--- a/taskcluster/ci/docker-image/kind.yml
+++ b/taskcluster/ci/docker-image/kind.yml
@@ -1,13 +1,13 @@
 # This Source Code Form is subject to the terms of the Mozilla Public
 # License, v. 2.0. If a copy of the MPL was not distributed with this
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
-implementation: 'taskgraph.kind.docker_image:DockerImageTask'
+implementation: 'taskgraph.kind.docker_image:DockerImageKind'
 images_path: '../../../testing/docker'
 
 # make a task for each docker-image we might want.  For the moment, since we
 # write artifacts for each, these are whitelisted, but ideally that will change
 # (to use subdirectory clones of the proper directory), at which point we can
 # generate tasks for every docker image in the directory, secure in the
 # knowledge that unnecessary images will be omitted from the target task graph
 images:

--- a/taskcluster/ci/legacy/kind.yml
+++ b/taskcluster/ci/legacy/kind.yml
@@ -1,6 +1,6 @@
 # This Source Code Form is subject to the terms of the Mozilla Public
 # License, v. 2.0. If a copy of the MPL was not distributed with this
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
-implementation: 'taskgraph.kind.legacy:LegacyTask'
+implementation: 'taskgraph.kind.legacy:LegacyKind'
 legacy_path: '.'

--- a/taskcluster/docs/taskgraph.rst
+++ b/taskcluster/docs/taskgraph.rst
@@ -36,29 +36,16 @@ differently.  Some kinds may generate ta
 example, symbol-upload tasks are all alike, and very simple), while other kinds
 may do little more than parse a directory of YAML files.
 
 A `kind.yml` file contains data about the kind, as well as referring to a
 Python class implementing the kind in its ``implementation`` key.  That
 implementation may rely on lots of code shared with other kinds, or contain a
 completely unique implementation of some functionality.
 
-The full list of pre-defined keys in this file is:
-
-``implementation``
-   Class implementing this kind, in the form ``<module-path>:<object-path>``.
-   This class should be a subclass of ``taskgraph.kind.base:Kind``.
-
-``kind-dependencies``
-   Kinds which should be loaded before this one.  This is useful when the kind
-   will use the list of already-created tasks to determine which tasks to
-   create, for example adding an upload-symbols task after every build task.
-
-Any other keys are subject to interpretation by the kind implementation.
-
 The result is a nice segmentation of implementation so that the more esoteric
 in-tree projects can do their crazy stuff in an isolated kind without making
 the bread-and-butter build and test configuration more complicated.
 
 Dependencies
 ------------
 
 Dependency links between tasks are always between different kinds(*).  At a

--- a/taskcluster/taskgraph/create.py
+++ b/taskcluster/taskgraph/create.py
@@ -44,18 +44,17 @@ def create_tasks(taskgraph, label_to_tas
             # task so that it does not start immediately; and so that if this loop
             # fails halfway through, none of the already-created tasks run.
             if decision_task_id and not task_def.get('dependencies'):
                 task_def['dependencies'] = [decision_task_id]
 
             task_def['taskGroupId'] = task_group_id
 
             # Wait for dependencies before submitting this.
-            deps_fs = [fs[dep] for dep in task_def.get('dependencies', [])
-                       if dep in fs]
+            deps_fs = [fs[dep] for dep in task_def['dependencies'] if dep in fs]
             for f in futures.as_completed(deps_fs):
                 f.result()
 
             fs[task_id] = e.submit(_create_task, session, task_id,
                                    taskid_to_label[task_id], task_def)
 
         # Wait for all futures to complete.
         for f in futures.as_completed(fs.values()):

--- a/taskcluster/taskgraph/generator.py
+++ b/taskcluster/taskgraph/generator.py
@@ -3,54 +3,22 @@
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
 from __future__ import absolute_import, print_function, unicode_literals
 import logging
 import os
 import yaml
 
 from .graph import Graph
-from .taskgraph import TaskGraph
+from .types import TaskGraph
 from .optimize import optimize_task_graph
 
 logger = logging.getLogger(__name__)
 
 
-class Kind(object):
-
-    def __init__(self, name, path, config):
-        self.name = name
-        self.path = path
-        self.config = config
-
-    def _get_impl_class(self):
-        # load the class defined by implementation
-        try:
-            impl = self.config['implementation']
-        except KeyError:
-            raise KeyError("{!r} does not define implementation".format(self.path))
-        if impl.count(':') != 1:
-            raise TypeError('{!r} implementation does not have the form "module:object"'
-                            .format(self.path))
-
-        impl_module, impl_object = impl.split(':')
-        impl_class = __import__(impl_module)
-        for a in impl_module.split('.')[1:]:
-            impl_class = getattr(impl_class, a)
-        for a in impl_object.split('.'):
-            impl_class = getattr(impl_class, a)
-
-        return impl_class
-
-    def load_tasks(self, parameters, loaded_tasks):
-        impl_class = self._get_impl_class()
-        return impl_class.load_tasks(self.name, self.path, self.config,
-                                     parameters, loaded_tasks)
-
-
 class TaskGraphGenerator(object):
     """
     The central controller for taskgraph.  This handles all phases of graph
     generation.  The task is generated from all of the kinds defined in
     subdirectories of the generator's root directory.
 
     Access to the results of this generation, as well as intermediate values at
     various phases of generation, is available via properties.  This encourages
@@ -142,52 +110,57 @@ class TaskGraphGenerator(object):
         """
         return self._run_until('label_to_taskid')
 
     def _load_kinds(self):
         for path in os.listdir(self.root_dir):
             path = os.path.join(self.root_dir, path)
             if not os.path.isdir(path):
                 continue
-            kind_name = os.path.basename(path)
-            logger.debug("loading kind `{}` from `{}`".format(kind_name, path))
+            name = os.path.basename(path)
+            logger.debug("loading kind `{}` from `{}`".format(name, path))
 
             kind_yml = os.path.join(path, 'kind.yml')
             with open(kind_yml) as f:
                 config = yaml.load(f)
 
-            yield Kind(kind_name, path, config)
+            # load the class defined by implementation
+            try:
+                impl = config['implementation']
+            except KeyError:
+                raise KeyError("{!r} does not define implementation".format(kind_yml))
+            if impl.count(':') != 1:
+                raise TypeError('{!r} implementation does not have the form "module:object"'
+                                .format(kind_yml))
+
+            impl_module, impl_object = impl.split(':')
+            impl_class = __import__(impl_module)
+            for a in impl_module.split('.')[1:]:
+                impl_class = getattr(impl_class, a)
+            for a in impl_object.split('.'):
+                impl_class = getattr(impl_class, a)
+
+            yield impl_class(path, config)
 
     def _run(self):
-        logger.info("Loading kinds")
-        # put the kinds into a graph and sort topologically so that kinds are loaded
-        # in post-order
-        kinds = {kind.name: kind for kind in self._load_kinds()}
-        edges = set()
-        for kind in kinds.itervalues():
-            for dep in kind.config.get('kind-dependencies', []):
-                edges.add((kind.name, dep, 'kind-dependency'))
-        kind_graph = Graph(set(kinds), edges)
-
         logger.info("Generating full task set")
         all_tasks = {}
-        for kind_name in kind_graph.visit_postorder():
-            logger.debug("Loading tasks for kind {}".format(kind_name))
-            kind = kinds[kind_name]
-            for task in kind.load_tasks(self.parameters, list(all_tasks.values())):
+        for kind in self._load_kinds():
+            for task in kind.load_tasks(self.parameters):
                 if task.label in all_tasks:
                     raise Exception("duplicate tasks with label " + task.label)
                 all_tasks[task.label] = task
+
         full_task_set = TaskGraph(all_tasks, Graph(set(all_tasks), set()))
         yield 'full_task_set', full_task_set
 
         logger.info("Generating full task graph")
         edges = set()
         for t in full_task_set:
-            for dep, depname in t.get_dependencies(full_task_set):
+            for dep, depname in t.kind.get_task_dependencies(t, full_task_set):
                 edges.add((t.label, dep, depname))
 
         full_task_graph = TaskGraph(all_tasks,
                                     Graph(full_task_set.graph.nodes, edges))
         yield 'full_task_graph', full_task_graph
 
         logger.info("Generating target task set")
         target_tasks = set(self.target_tasks_method(full_task_graph, self.parameters))

--- a/taskcluster/taskgraph/kind/base.py
+++ b/taskcluster/taskgraph/kind/base.py
@@ -1,87 +1,53 @@
 # This Source Code Form is subject to the terms of the Mozilla Public
 # License, v. 2.0. If a copy of the MPL was not distributed with this
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
 from __future__ import absolute_import, print_function, unicode_literals
 
+import os
 import abc
 
 
-class Task(object):
+class Kind(object):
     """
-    Representation of a task in a TaskGraph.  Each Task has, at creation:
-
-    - kind: the name of the task kind
-    - label; the label for this task
-    - attributes: a dictionary of attributes for this task (used for filtering)
-    - task: the task definition (JSON-able dictionary)
-
-    And later, as the task-graph processing proceeds:
-
-    - task_id -- TaskCluster taskId under which this task will be created
-    - optimized -- true if this task need not be performed
-
     A kind represents a collection of tasks that share common characteristics.
     For example, all build jobs.  Each instance of a kind is intialized with a
     path from which it draws its task configuration.  The instance is free to
     store as much local state as it needs.
     """
     __metaclass__ = abc.ABCMeta
 
-    def __init__(self, kind, label, attributes, task):
-        self.kind = kind
-        self.label = label
-        self.attributes = attributes
-        self.task = task
-
-        self.task_id = None
-        self.optimized = False
-
-        self.attributes['kind'] = kind
+    def __init__(self, path, config):
+        self.name = os.path.basename(path)
+        self.path = path
+        self.config = config
 
-        if not (all(isinstance(x, basestring) for x in self.attributes.iterkeys()) and
-                all(isinstance(x, basestring) for x in self.attributes.itervalues())):
-            raise TypeError("attribute names and values must be strings")
-
-    @classmethod
     @abc.abstractmethod
-    def load_tasks(cls, kind, path, config, parameters, loaded_tasks):
+    def load_tasks(self, parameters):
         """
-        Load the tasks for a given kind.
-
-        The `kind` is the name of the kind; the configuration for that kind
-        named this class.
-
-        The `path` is the path to the configuration directory for the kind.  This
-        can be used to load extra data, templates, etc.
+        Get the set of tasks of this kind.
 
         The `parameters` give details on which to base the task generation.
         See `taskcluster/docs/parameters.rst` for details.
 
-        At the time this method is called, all kinds on which this kind depends
-        (that is, specified in the `kind-dependencies` key in `self.config`
-        have already loaded their tasks, and those tasks are available in
-        the list `loaded_tasks`.
-
         The return value is a list of Task instances.
         """
 
     @abc.abstractmethod
-    def get_dependencies(self, taskgraph):
+    def get_task_dependencies(self, task, taskgraph):
         """
-        Get the set of task labels this task depends on, by querying the full
-        task set, given as `taskgraph`.
+        Get the set of task labels this task depends on, by querying the task graph.
 
         Returns a list of (task_label, dependency_name) pairs describing the
         dependencies.
         """
 
-    def optimize(self):
+    def optimize_task(self, task):
         """
         Determine whether this task can be optimized, and if it can, what taskId
         it should be replaced with.
 
         The return value is a tuple `(optimized, taskId)`.  If `optimized` is
         true, then the task will be optimized (in other words, not included in
         the task graph).  If the second argument is a taskid, then any
         dependencies on this task will isntead depend on that taskId.  It is an

--- a/taskcluster/taskgraph/kind/docker_image.py
+++ b/taskcluster/taskgraph/kind/docker_image.py
@@ -7,40 +7,36 @@ from __future__ import absolute_import, 
 import logging
 import json
 import os
 import urllib2
 import tarfile
 import time
 
 from . import base
+from ..types import Task
 from taskgraph.util.docker import (
     docker_image,
     generate_context_hash
 )
 from taskgraph.util.templates import Templates
 from taskgraph.util.time import (
     json_time_from_now,
     current_json_time,
 )
 
 logger = logging.getLogger(__name__)
 GECKO = os.path.realpath(os.path.join(__file__, '..', '..', '..', '..'))
 ARTIFACT_URL = 'https://queue.taskcluster.net/v1/task/{}/artifacts/{}'
 INDEX_URL = 'https://index.taskcluster.net/v1/task/{}'
 
 
-class DockerImageTask(base.Task):
+class DockerImageKind(base.Kind):
 
-    def __init__(self, *args, **kwargs):
-        self.index_paths = kwargs.pop('index_paths')
-        super(DockerImageTask, self).__init__(*args, **kwargs)
-
-    @classmethod
-    def load_tasks(cls, kind, path, config, params, loaded_tasks):
+    def load_tasks(self, params):
         # TODO: make this match the pushdate (get it from a parameter rather than vcs)
         pushdate = time.strftime('%Y%m%d%H%M%S', time.gmtime())
 
         parameters = {
             'pushlog_id': params.get('pushlog_id', 0),
             'pushdate': pushdate,
             'pushtime': pushdate[8:],
             'year': pushdate[0:4],
@@ -56,18 +52,18 @@ class DockerImageTask(base.Task):
             'level': params['level'],
             'from_now': json_time_from_now,
             'now': current_json_time(),
             'source': '{repo}file/{rev}/testing/taskcluster/tasks/image.yml'
                       .format(repo=params['head_repository'], rev=params['head_rev']),
         }
 
         tasks = []
-        templates = Templates(path)
-        for image_name in config['images']:
+        templates = Templates(self.path)
+        for image_name in self.config['images']:
             context_path = os.path.join('testing', 'docker', image_name)
             context_hash = generate_context_hash(context_path)
 
             image_parameters = dict(parameters)
             image_parameters['context_hash'] = context_hash
             image_parameters['context_path'] = context_path
             image_parameters['artifact_path'] = 'public/image.tar'
             image_parameters['image_name'] = image_name
@@ -75,47 +71,50 @@ class DockerImageTask(base.Task):
             image_artifact_path = \
                 "public/decision_task/image_contexts/{}/context.tar.gz".format(image_name)
             if os.environ.get('TASK_ID'):
                 destination = os.path.join(
                     os.environ['HOME'],
                     "artifacts/decision_task/image_contexts/{}/context.tar.gz".format(image_name))
                 image_parameters['context_url'] = ARTIFACT_URL.format(
                     os.environ['TASK_ID'], image_artifact_path)
-                cls.create_context_tar(context_path, destination, image_name)
+                self.create_context_tar(context_path, destination, image_name)
             else:
                 # skip context generation since this isn't a decision task
                 # TODO: generate context tarballs using subdirectory clones in
                 # the image-building task so we don't have to worry about this.
                 image_parameters['context_url'] = 'file:///tmp/' + image_artifact_path
 
             image_task = templates.load('image.yml', image_parameters)
 
-            attributes = {'image_name': image_name}
+            attributes = {
+                'kind': self.name,
+                'image_name': image_name,
+            }
 
             # As an optimization, if the context hash exists for mozilla-central, that image
             # task ID will be used.  The reasoning behind this is that eventually everything ends
             # up on mozilla-central at some point if most tasks use this as a common image
             # for a given context hash, a worker within Taskcluster does not need to contain
             # the same image per branch.
             index_paths = ['docker.images.v1.{}.{}.hash.{}'.format(
                                 project, image_name, context_hash)
                            for project in ['mozilla-central', params['project']]]
 
-            tasks.append(cls(kind, 'build-docker-image-' + image_name,
-                             task=image_task['task'], attributes=attributes,
-                             index_paths=index_paths))
+            tasks.append(Task(self, 'build-docker-image-' + image_name,
+                              task=image_task['task'], attributes=attributes,
+                              index_paths=index_paths))
 
         return tasks
 
-    def get_dependencies(self, taskgraph):
+    def get_task_dependencies(self, task, taskgraph):
         return []
 
-    def optimize(self):
-        for index_path in self.index_paths:
+    def optimize_task(self, task, taskgraph):
+        for index_path in task.extra['index_paths']:
             try:
                 url = INDEX_URL.format(index_path)
                 existing_task = json.load(urllib2.urlopen(url))
                 # Only return the task ID if the artifact exists for the indexed
                 # task.  Otherwise, continue on looking at each of the branches.  Method
                 # continues trying other branches in case mozilla-central has an expired
                 # artifact, but 'project' might not. Only return no task ID if all
                 # branches have been tried
@@ -126,17 +125,16 @@ class DockerImageTask(base.Task):
 
                 # HEAD success on the artifact is enough
                 return True, existing_task['taskId']
             except urllib2.HTTPError:
                 pass
 
         return False, None
 
-    @classmethod
-    def create_context_tar(cls, context_dir, destination, image_name):
+    def create_context_tar(self, context_dir, destination, image_name):
         'Creates a tar file of a particular context directory.'
         destination = os.path.abspath(destination)
         if not os.path.exists(os.path.dirname(destination)):
             os.makedirs(os.path.dirname(destination))
 
         with tarfile.open(destination, 'w:gz') as tar:
             tar.add(context_dir, arcname=image_name)

--- a/taskcluster/taskgraph/kind/legacy.py
+++ b/taskcluster/taskgraph/kind/legacy.py
@@ -8,16 +8,17 @@ import copy
 import json
 import logging
 import os
 import re
 import time
 from collections import namedtuple
 
 from . import base
+from ..types import Task
 from mozpack.path import match as mozpackmatch
 from slugid import nice as slugid
 from taskgraph.util.legacy_commit_parser import parse_commit
 from taskgraph.util.time import (
     json_time_from_now,
     current_json_time,
 )
 from taskgraph.util.templates import Templates
@@ -65,20 +66,20 @@ def gaia_info():
 
     if gaia['git'] is None or \
        gaia['git']['remote'] == '' or \
        gaia['git']['git_revision'] == '' or \
        gaia['git']['branch'] == '':
 
         # Just use the hg params...
         return {
-            'gaia_base_repository': 'https://hg.mozilla.org/{}'.format(gaia['repo_path']),
-            'gaia_head_repository': 'https://hg.mozilla.org/{}'.format(gaia['repo_path']),
-            'gaia_ref': gaia['revision'],
-            'gaia_rev': gaia['revision']
+          'gaia_base_repository': 'https://hg.mozilla.org/{}'.format(gaia['repo_path']),
+          'gaia_head_repository': 'https://hg.mozilla.org/{}'.format(gaia['repo_path']),
+          'gaia_ref': gaia['revision'],
+          'gaia_rev': gaia['revision']
         }
 
     else:
         # Use git
         return {
             'gaia_base_repository': gaia['git']['remote'],
             'gaia_head_repository': gaia['git']['remote'],
             'gaia_rev': gaia['git']['git_revision'],
@@ -286,33 +287,28 @@ def validate_build_task(task):
         if 'build' not in locations:
             raise BuildTaskValidationException('task.extra.locations.build missing')
 
         if 'tests' not in locations and 'test_packages' not in locations:
             raise BuildTaskValidationException('task.extra.locations.tests or '
                                                'task.extra.locations.tests_packages missing')
 
 
-class LegacyTask(base.Task):
+class LegacyKind(base.Kind):
     """
     This kind generates a full task graph from the old YAML files in
     `testing/taskcluster/tasks`.  The tasks already have dependency links.
 
     The existing task-graph generation generates slugids for tasks during task
     generation, so this kind labels tasks using those slugids, with a prefix of
     "TaskLabel==".  These labels are unfortunately not stable from run to run.
     """
 
-    def __init__(self, *args, **kwargs):
-        self.task_dict = kwargs.pop('task_dict')
-        super(LegacyTask, self).__init__(*args, **kwargs)
-
-    @classmethod
-    def load_tasks(cls, kind, path, config, params, loaded_tasks):
-        root = os.path.abspath(os.path.join(path, config['legacy_path']))
+    def load_tasks(self, params):
+        root = os.path.abspath(os.path.join(self.path, self.config['legacy_path']))
 
         project = params['project']
         # NOTE: message is ignored here; we always use DEFAULT_TRY, then filter the
         # resulting task graph later
         message = DEFAULT_TRY
 
         templates = Templates(root)
 
@@ -382,17 +378,17 @@ class LegacyTask(base.Task):
             route = format_treeherder_route(TREEHERDER_ROUTES[env],
                                             parameters['project'],
                                             parameters['head_rev'],
                                             parameters['pushlog_id'])
             graph['scopes'].add("queue:route:{}".format(route))
 
         graph['metadata'] = {
             'source': '{repo}file/{rev}/testing/taskcluster/mach_commands.py'.format(
-                repo=params['head_repository'], rev=params['head_rev']),
+                        repo=params['head_repository'], rev=params['head_rev']),
             'owner': params['owner'],
             # TODO: Add full mach commands to this example?
             'description': 'Task graph generated via ./mach taskcluster-graph',
             'name': 'task graph local'
         }
 
         # Filter the job graph according to conditions met by this invocation run.
         def should_run(task):
@@ -496,17 +492,17 @@ class LegacyTask(base.Task):
                             all_routes[route],
                         ))
                 all_routes[route] = build_task['task']['metadata']['name']
 
             graph['scopes'].add(define_task)
             graph['scopes'] |= set(build_task['task'].get('scopes', []))
             route_scopes = map(
                 lambda route: 'queue:route:' + route, build_task['task'].get('routes', [])
-            )
+                )
             graph['scopes'] |= set(route_scopes)
 
             # Treeherder symbol configuration for the graph required for each
             # build so tests know which platform they belong to.
             build_treeherder_config = build_task['task']['extra']['treeherder']
 
             if 'machine' not in build_treeherder_config:
                 message = '({}), extra.treeherder.machine required for all builds'
@@ -612,30 +608,33 @@ class LegacyTask(base.Task):
                         test_task['task']['workerType']
                     )
 
                     graph['scopes'].add(define_task)
                     graph['scopes'] |= set(test_task['task'].get('scopes', []))
 
         graph['scopes'] = sorted(graph['scopes'])
 
+        # save the graph for later, when taskgraph asks for additional information
+        # such as dependencies
+        self.graph = graph
+        self.tasks_by_label = {t['taskId']: t for t in self.graph['tasks']}
+
         # Convert to a dictionary of tasks.  The process above has invented a
         # taskId for each task, and we use those as the *labels* for the tasks;
         # taskgraph will later assign them new taskIds.
-        return [
-            cls(kind, t['taskId'], task=t['task'], attributes=t['attributes'], task_dict=t)
-            for t in graph['tasks']
-        ]
+        return [Task(self, t['taskId'], task=t['task'], attributes=t['attributes'])
+                for t in self.graph['tasks']]
 
-    def get_dependencies(self, taskgraph):
+    def get_task_dependencies(self, task, taskgraph):
         # fetch dependency information from the cached graph
-        deps = [(label, label) for label in self.task_dict.get('requires', [])]
+        taskdict = self.tasks_by_label[task.label]
+        deps = [(label, label) for label in taskdict.get('requires', [])]
 
         # add a dependency on an image task, if needed
-        if 'docker-image' in self.task_dict:
-            deps.append(('build-docker-image-{docker-image}'.format(**self.task_dict),
-                         'docker-image'))
+        if 'docker-image' in taskdict:
+            deps.append(('build-docker-image-{docker-image}'.format(**taskdict), 'docker-image'))
 
         return deps
 
-    def optimize(self, taskgraph):
+    def optimize_task(self, task, taskgraph):
         # no optimization for the moment
         return False, None

--- a/taskcluster/taskgraph/optimize.py
+++ b/taskcluster/taskgraph/optimize.py
@@ -2,17 +2,17 @@
 # License, v. 2.0. If a copy of the MPL was not distributed with this
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
 from __future__ import absolute_import, print_function, unicode_literals
 import logging
 import re
 
 from .graph import Graph
-from .taskgraph import TaskGraph
+from .types import TaskGraph
 from slugid import nice as slugid
 
 logger = logging.getLogger(__name__)
 TASK_REFERENCE_PATTERN = re.compile('<([^>]+)>')
 
 
 def optimize_task_graph(target_task_graph, do_not_optimize):
     """
@@ -83,17 +83,17 @@ def annotate_task_graph(target_task_grap
         replacement_task_id = None
         if label in do_not_optimize:
             optimized = False
         # if any dependencies can't be optimized, this task can't, either
         elif any(not t.optimized for t in dependencies):
             optimized = False
         # otherwise, examine the task itself (which may be an expensive operation)
         else:
-            optimized, replacement_task_id = task.optimize()
+            optimized, replacement_task_id = task.kind.optimize_task(task, named_task_dependencies)
 
         task.optimized = optimized
         task.task_id = replacement_task_id
         if replacement_task_id:
             label_to_taskid[label] = replacement_task_id
 
         if optimized:
             if replacement_task_id:

--- a/taskcluster/taskgraph/test/test_create.py
+++ b/taskcluster/taskgraph/test/test_create.py
@@ -4,22 +4,30 @@
 
 from __future__ import absolute_import, print_function, unicode_literals
 
 import unittest
 import os
 
 from .. import create
 from ..graph import Graph
-from ..taskgraph import TaskGraph
-from .util import TestTask
+from ..types import Task, TaskGraph
 
 from mozunit import main
 
 
+class FakeKind(object):
+
+    def get_task_definition(self, task, deps_by_name):
+        # sanity-check the deps_by_name
+        for k, v in deps_by_name.iteritems():
+            assert k == 'edge'
+        return {'payload': 'hello world'}
+
+
 class TestCreate(unittest.TestCase):
 
     def setUp(self):
         self.old_task_id = os.environ.get('TASK_ID')
         if 'TASK_ID' in os.environ:
             del os.environ['TASK_ID']
         self.created_tasks = {}
         self.old_create_task = create._create_task
@@ -31,19 +39,21 @@ class TestCreate(unittest.TestCase):
             os.environ['TASK_ID'] = self.old_task_id
         elif 'TASK_ID' in os.environ:
             del os.environ['TASK_ID']
 
     def fake_create_task(self, session, task_id, label, task_def):
         self.created_tasks[task_id] = task_def
 
     def test_create_tasks(self):
+        os.environ['TASK_ID'] = 'decisiontask'
+        kind = FakeKind()
         tasks = {
-            'tid-a': TestTask(label='a', task={'payload': 'hello world'}),
-            'tid-b': TestTask(label='b', task={'payload': 'hello world'}),
+            'tid-a': Task(kind=kind, label='a', task={'payload': 'hello world'}),
+            'tid-b': Task(kind=kind, label='b', task={'payload': 'hello world'}),
         }
         label_to_taskid = {'a': 'tid-a', 'b': 'tid-b'}
         graph = Graph(nodes={'tid-a', 'tid-b'}, edges={('tid-a', 'tid-b', 'edge')})
         taskgraph = TaskGraph(tasks, graph)
 
         create.create_tasks(taskgraph, label_to_taskid)
 
         for tid, task in self.created_tasks.iteritems():
@@ -53,23 +63,24 @@ class TestCreate(unittest.TestCase):
                 if depid is 'decisiontask':
                     # Don't look for decisiontask here
                     continue
                 self.assertIn(depid, self.created_tasks)
 
     def test_create_task_without_dependencies(self):
         "a task with no dependencies depends on the decision task"
         os.environ['TASK_ID'] = 'decisiontask'
+        kind = FakeKind()
         tasks = {
-            'tid-a': TestTask(label='a', task={'payload': 'hello world'}),
+            'tid-a': Task(kind=kind, label='a', task={'payload': 'hello world'}),
         }
         label_to_taskid = {'a': 'tid-a'}
         graph = Graph(nodes={'tid-a'}, edges=set())
         taskgraph = TaskGraph(tasks, graph)
 
         create.create_tasks(taskgraph, label_to_taskid)
 
         for tid, task in self.created_tasks.iteritems():
-            self.assertEqual(task.get('dependencies'), [os.environ['TASK_ID']])
+            self.assertEqual(task['dependencies'], [os.environ['TASK_ID']])
 
 
 if __name__ == '__main__':
     main()

--- a/taskcluster/taskgraph/test/test_decision.py
+++ b/taskcluster/taskgraph/test/test_decision.py
@@ -8,43 +8,42 @@ import os
 import json
 import yaml
 import shutil
 import unittest
 import tempfile
 
 from .. import decision
 from ..graph import Graph
-from ..taskgraph import TaskGraph
-from .util import TestTask
+from ..types import Task, TaskGraph
 from mozunit import main
 
 
 class TestDecision(unittest.TestCase):
 
     def test_taskgraph_to_json(self):
         tasks = {
-            'a': TestTask(label='a', attributes={'attr': 'a-task'}),
-            'b': TestTask(label='b', task={'task': 'def'}),
+            'a': Task(kind=None, label='a', attributes={'attr': 'a-task'}),
+            'b': Task(kind=None, label='b', task={'task': 'def'}),
         }
         graph = Graph(nodes=set('ab'), edges={('a', 'b', 'edgelabel')})
         taskgraph = TaskGraph(tasks, graph)
 
         res = taskgraph.to_json()
 
         self.assertEqual(res, {
             'a': {
                 'label': 'a',
-                'attributes': {'attr': 'a-task', 'kind': 'test'},
+                'attributes': {'attr': 'a-task'},
                 'task': {},
                 'dependencies': {'edgelabel': 'b'},
             },
             'b': {
                 'label': 'b',
-                'attributes': {'kind': 'test'},
+                'attributes': {},
                 'task': {'task': 'def'},
                 'dependencies': {},
             }
         })
 
     def test_write_artifact_json(self):
         data = [{'some': 'data'}]
         tmpdir = tempfile.mkdtemp()

--- a/taskcluster/taskgraph/test/test_generator.py
+++ b/taskcluster/taskgraph/test/test_generator.py
@@ -1,129 +1,103 @@
 # This Source Code Form is subject to the terms of the Mozilla Public
 # License, v. 2.0. If a copy of the MPL was not distributed with this
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
 from __future__ import absolute_import, print_function, unicode_literals
 
 import unittest
 
-from ..generator import TaskGraphGenerator, Kind
+from ..generator import TaskGraphGenerator
+from .. import types
 from .. import graph
-from ..kind import base
 from mozunit import main
 
 
-class FakeTask(base.Task):
-
-    def __init__(self, **kwargs):
-        self.i = kwargs.pop('i')
-        super(FakeTask, self).__init__(**kwargs)
+class FakeKind(object):
 
-    @classmethod
-    def load_tasks(cls, kind, path, config, parameters, loaded_tasks):
-        return [cls(kind=kind,
-                    label='{}-t-{}'.format(kind, i),
-                    attributes={'tasknum': str(i)},
-                    task={},
-                    i=i)
-                for i in range(3)]
+    def maketask(self, i):
+        return types.Task(
+            self,
+            label='t-{}'.format(i),
+            attributes={'tasknum': str(i)},
+            task={},
+            i=i)
 
-    def get_dependencies(self, full_task_set):
-        i = self.i
+    def load_tasks(self, parameters):
+        self.tasks = [self.maketask(i) for i in range(3)]
+        return self.tasks
+
+    def get_task_dependencies(self, task, full_task_set):
+        i = task.extra['i']
         if i > 0:
-            return [('{}-t-{}'.format(self.kind, i - 1), 'prev')]
+            return [('t-{}'.format(i - 1), 'prev')]
         else:
             return []
 
-    def optimize(self):
+    def optimize_task(self, task, dependencies):
         return False, None
 
 
-class FakeKind(Kind):
-
-    def _get_impl_class(self):
-        return FakeTask
-
-    def load_tasks(self, parameters, loaded_tasks):
-        FakeKind.loaded_kinds.append(self.name)
-        return super(FakeKind, self).load_tasks(parameters, loaded_tasks)
-
-
 class WithFakeKind(TaskGraphGenerator):
 
     def _load_kinds(self):
-        for kind_name, deps in self.parameters['kinds']:
-            yield FakeKind(
-                kind_name, '/fake',
-                {'kind-dependencies': deps} if deps else {})
+        yield FakeKind()
 
 
 class TestGenerator(unittest.TestCase):
 
-    def maketgg(self, target_tasks=None, kinds=[('fake', [])]):
-        FakeKind.loaded_kinds = []
-        self.target_tasks = target_tasks or []
+    def setUp(self):
+        self.target_tasks = []
 
         def target_tasks_method(full_task_graph, parameters):
             return self.target_tasks
-        return WithFakeKind('/root', {'kinds': kinds}, target_tasks_method)
-
-    def test_kind_ordering(self):
-        "When task kinds depend on each other, they are loaded in postorder"
-        self.tgg = self.maketgg(kinds=[
-            ('fake3', ['fake2', 'fake1']),
-            ('fake2', ['fake1']),
-            ('fake1', []),
-        ])
-        self.tgg._run_until('full_task_set')
-        self.assertEqual(FakeKind.loaded_kinds, ['fake1', 'fake2', 'fake3'])
+        self.tgg = WithFakeKind('/root', {}, target_tasks_method)
 
     def test_full_task_set(self):
         "The full_task_set property has all tasks"
-        self.tgg = self.maketgg()
         self.assertEqual(self.tgg.full_task_set.graph,
-                         graph.Graph({'fake-t-0', 'fake-t-1', 'fake-t-2'}, set()))
-        self.assertEqual(sorted(self.tgg.full_task_set.tasks.keys()),
-                         sorted(['fake-t-0', 'fake-t-1', 'fake-t-2']))
+                         graph.Graph({'t-0', 't-1', 't-2'}, set()))
+        self.assertEqual(self.tgg.full_task_set.tasks.keys(),
+                         ['t-0', 't-1', 't-2'])
 
     def test_full_task_graph(self):
         "The full_task_graph property has all tasks, and links"
-        self.tgg = self.maketgg()
         self.assertEqual(self.tgg.full_task_graph.graph,
-                         graph.Graph({'fake-t-0', 'fake-t-1', 'fake-t-2'},
+                         graph.Graph({'t-0', 't-1', 't-2'},
                                      {
-                                         ('fake-t-1', 'fake-t-0', 'prev'),
-                                         ('fake-t-2', 'fake-t-1', 'prev'),
+                                         ('t-1', 't-0', 'prev'),
+                                         ('t-2', 't-1', 'prev'),
                          }))
-        self.assertEqual(sorted(self.tgg.full_task_graph.tasks.keys()),
-                         sorted(['fake-t-0', 'fake-t-1', 'fake-t-2']))
+        self.assertEqual(self.tgg.full_task_graph.tasks.keys(),
+                         ['t-0', 't-1', 't-2'])
 
     def test_target_task_set(self):
         "The target_task_set property has the targeted tasks"
-        self.tgg = self.maketgg(['fake-t-1'])
+        self.target_tasks = ['t-1']
         self.assertEqual(self.tgg.target_task_set.graph,
-                         graph.Graph({'fake-t-1'}, set()))
+                         graph.Graph({'t-1'}, set()))
         self.assertEqual(self.tgg.target_task_set.tasks.keys(),
-                         ['fake-t-1'])
+                         ['t-1'])
 
     def test_target_task_graph(self):
         "The target_task_graph property has the targeted tasks and deps"
-        self.tgg = self.maketgg(['fake-t-1'])
+        self.target_tasks = ['t-1']
         self.assertEqual(self.tgg.target_task_graph.graph,
-                         graph.Graph({'fake-t-0', 'fake-t-1'},
-                                     {('fake-t-1', 'fake-t-0', 'prev')}))
+                         graph.Graph({'t-0', 't-1'},
+                                     {('t-1', 't-0', 'prev')}))
         self.assertEqual(sorted(self.tgg.target_task_graph.tasks.keys()),
-                         sorted(['fake-t-0', 'fake-t-1']))
+                         sorted(['t-0', 't-1']))
 
     def test_optimized_task_graph(self):
         "The optimized task graph contains task ids"
-        self.tgg = self.maketgg(['fake-t-2'])
+        self.target_tasks = ['t-2']
         tid = self.tgg.label_to_taskid
         self.assertEqual(
             self.tgg.optimized_task_graph.graph,
-            graph.Graph({tid['fake-t-0'], tid['fake-t-1'], tid['fake-t-2']}, {
-                (tid['fake-t-1'], tid['fake-t-0'], 'prev'),
-                (tid['fake-t-2'], tid['fake-t-1'], 'prev'),
-            }))
+            graph.Graph({tid['t-0'], tid['t-1'], tid['t-2']}, {
+                (tid['t-1'], tid['t-0'], 'prev'),
+                (tid['t-2'], tid['t-1'], 'prev'),
+            })
+            )
 
 if __name__ == '__main__':
     main()

--- a/taskcluster/taskgraph/test/test_kind_docker_image.py
+++ b/taskcluster/taskgraph/test/test_kind_docker_image.py
@@ -7,36 +7,30 @@ from __future__ import absolute_import, 
 import unittest
 import tempfile
 import os
 
 from ..kind import docker_image
 from mozunit import main
 
 
-KIND_PATH = os.path.join(docker_image.GECKO, 'taskcluster', 'ci', 'docker-image')
-
-
 class TestDockerImageKind(unittest.TestCase):
 
     def setUp(self):
-        self.task = docker_image.DockerImageTask(
-            'docker-image',
-            KIND_PATH,
-            {},
-            {},
-            index_paths=[])
+        self.kind = docker_image.DockerImageKind(
+                os.path.join(docker_image.GECKO, 'taskcluster', 'ci', 'docker-image'),
+                {})
 
     def test_get_task_dependencies(self):
         # this one's easy!
-        self.assertEqual(self.task.get_dependencies(None), [])
+        self.assertEqual(self.kind.get_task_dependencies(None, None), [])
 
     # TODO: optimize_task
 
     def test_create_context_tar(self):
         image_dir = os.path.join(docker_image.GECKO, 'testing', 'docker', 'image_builder')
         tarball = tempfile.mkstemp()[1]
-        self.task.create_context_tar(image_dir, tarball, 'image_builder')
+        self.kind.create_context_tar(image_dir, tarball, 'image_builder')
         self.failUnless(os.path.exists(tarball))
         os.unlink(tarball)
 
 if __name__ == '__main__':
     main()

--- a/taskcluster/taskgraph/test/test_kind_legacy.py
+++ b/taskcluster/taskgraph/test/test_kind_legacy.py
@@ -2,22 +2,32 @@
 # License, v. 2.0. If a copy of the MPL was not distributed with this
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
 from __future__ import absolute_import, print_function, unicode_literals
 
 import unittest
 
 from ..kind.legacy import (
+    LegacyKind,
     validate_build_task,
     BuildTaskValidationException
 )
 from mozunit import main
 
 
+class TestLegacyKind(unittest.TestCase):
+    # NOTE: much of LegacyKind is copy-pasted from the old legacy code, which
+    # is emphatically *not* designed for testing, so this test class does not
+    # attempt to test the entire class.
+
+    def setUp(self):
+        self.kind = LegacyKind('/root', {})
+
+
 class TestValidateBuildTask(unittest.TestCase):
 
     def test_validate_missing_extra(self):
         with self.assertRaises(BuildTaskValidationException):
             validate_build_task({})
 
     def test_validate_valid(self):
         with self.assertRaises(BuildTaskValidationException):

--- a/taskcluster/taskgraph/test/test_optimize.py
+++ b/taskcluster/taskgraph/test/test_optimize.py
@@ -3,19 +3,18 @@
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
 from __future__ import absolute_import, print_function, unicode_literals
 
 import unittest
 
 from ..optimize import optimize_task_graph, resolve_task_references
 from ..optimize import annotate_task_graph, get_subgraph
-from ..taskgraph import TaskGraph
+from .. import types
 from .. import graph
-from .util import TestTask
 
 
 class TestResolveTaskReferences(unittest.TestCase):
 
     def do(self, input, output):
         taskid_for_edge_name = {'edge%d' % n: 'tid%d' % n for n in range(1, 4)}
         self.assertEqual(resolve_task_references('subject', input, taskid_for_edge_name), output)
 
@@ -45,127 +44,130 @@ class TestResolveTaskReferences(unittest
                 {'escape': '<tid3>'})
 
     def test_invalid(self):
         "resolve_task_references raises a KeyError on reference to an invalid task"
         self.assertRaisesRegexp(
             KeyError,
             "task 'subject' has no dependency with label 'no-such'",
             lambda: resolve_task_references('subject', {'task-reference': '<no-such>'}, {})
-        )
+            )
 
 
-class OptimizingTask(TestTask):
-    # the `optimize` method on this class is overridden direclty in the tests
-    # below.
-    pass
+class FakeKind(object):
+
+    def __init__(self, optimize_task):
+        self.optimize_task = optimize_task
 
 
 class TestOptimize(unittest.TestCase):
 
     kind = None
 
+    def make_kind(self, optimize_task):
+        self.kind = FakeKind(optimize_task)
+
     def make_task(self, label, task_def=None, optimized=None, task_id=None):
         task_def = task_def or {'sample': 'task-def'}
-        task = OptimizingTask(label=label, task=task_def)
+        task = types.Task(self.kind, label=label, task=task_def)
         task.optimized = optimized
         task.task_id = task_id
         return task
 
     def make_graph(self, *tasks_and_edges):
-        tasks = {t.label: t for t in tasks_and_edges if isinstance(t, OptimizingTask)}
-        edges = {e for e in tasks_and_edges if not isinstance(e, OptimizingTask)}
-        return TaskGraph(tasks, graph.Graph(set(tasks), edges))
+        tasks = {t.label: t for t in tasks_and_edges if isinstance(t, types.Task)}
+        edges = {e for e in tasks_and_edges if not isinstance(e, types.Task)}
+        return types.TaskGraph(tasks, graph.Graph(set(tasks), edges))
 
     def assert_annotations(self, graph, **annotations):
         def repl(task_id):
             return 'SLUGID' if task_id and len(task_id) == 22 else task_id
         got_annotations = {
             t.label: (t.optimized, repl(t.task_id)) for t in graph.tasks.itervalues()
-        }
+            }
         self.assertEqual(got_annotations, annotations)
 
     def test_annotate_task_graph_no_optimize(self):
         "annotating marks everything as un-optimized if the kind returns that"
-        OptimizingTask.optimize = lambda self: (False, None)
+        self.make_kind(lambda task, deps: (False, None))
         graph = self.make_graph(
             self.make_task('task1'),
             self.make_task('task2'),
             self.make_task('task3'),
             ('task2', 'task1', 'build'),
             ('task2', 'task3', 'image'),
         )
         annotate_task_graph(graph, set(), graph.graph.named_links_dict(), {})
         self.assert_annotations(
             graph,
             task1=(False, None),
             task2=(False, None),
             task3=(False, None)
-        )
+            )
 
     def test_annotate_task_graph_taskid_without_optimize(self):
         "raises exception if kind returns a taskid without optimizing"
-        OptimizingTask.optimize = lambda self: (False, 'some-taskid')
+        self.make_kind(lambda task, deps: (False, 'some-taskid'))
         graph = self.make_graph(self.make_task('task1'))
         self.assertRaises(
             Exception,
             lambda: annotate_task_graph(graph, set(), graph.graph.named_links_dict(), {})
-        )
+            )
 
     def test_annotate_task_graph_optimize_away_dependency(self):
         "raises exception if kind optimizes away a task on which another depends"
-        OptimizingTask.optimize = \
-            lambda self: (True, None) if self.label == 'task1' else (False, None)
+        self.make_kind(lambda task, deps: (True, None) if task.label == 'task1' else (False, None))
         graph = self.make_graph(
             self.make_task('task1'),
             self.make_task('task2'),
             ('task2', 'task1', 'build'),
         )
         self.assertRaises(
             Exception,
             lambda: annotate_task_graph(graph, set(), graph.graph.named_links_dict(), {})
-        )
+            )
 
     def test_annotate_task_graph_do_not_optimize(self):
         "annotating marks everything as un-optimized if in do_not_optimize"
-        OptimizingTask.optimize = lambda self: (True, 'taskid')
+        self.make_kind(lambda task, deps: (True, 'taskid'))
         graph = self.make_graph(
             self.make_task('task1'),
             self.make_task('task2'),
             ('task2', 'task1', 'build'),
         )
         label_to_taskid = {}
         annotate_task_graph(graph, {'task1', 'task2'},
                             graph.graph.named_links_dict(), label_to_taskid)
         self.assert_annotations(
             graph,
             task1=(False, None),
             task2=(False, None)
-        )
+            )
         self.assertEqual
 
     def test_annotate_task_graph_nos_propagate(self):
         "annotating marks a task with a non-optimized dependency as non-optimized"
-        OptimizingTask.optimize = \
-            lambda self: (False, None) if self.label == 'task1' else (True, 'taskid')
+        self.make_kind(
+            lambda task, deps: (False, None) if task.label == 'task1' else (True, 'taskid')
+            )
         graph = self.make_graph(
             self.make_task('task1'),
             self.make_task('task2'),
             self.make_task('task3'),
             ('task2', 'task1', 'build'),
             ('task2', 'task3', 'image'),
         )
         annotate_task_graph(graph, set(),
                             graph.graph.named_links_dict(), {})
         self.assert_annotations(
             graph,
             task1=(False, None),
             task2=(False, None),  # kind would have returned (True, 'taskid') here
             task3=(True, 'taskid')
-        )
+            )
 
     def test_get_subgraph_single_dep(self):
         "when a single dependency is optimized, it is omitted from the graph"
         graph = self.make_graph(
             self.make_task('task1', optimized=True, task_id='dep1'),
             self.make_task('task2', optimized=False),
             self.make_task('task3', optimized=False),
             ('task2', 'task1', 'build'),
@@ -218,17 +220,17 @@ class TestOptimize(unittest.TestCase):
     def test_get_subgraph_refs_resolved(self):
         "get_subgraph resolves task references"
         graph = self.make_graph(
             self.make_task('task1', optimized=True, task_id='dep1'),
             self.make_task(
                 'task2',
                 optimized=False,
                 task_def={'payload': {'task-reference': 'http://<build>/<test>'}}
-            ),
+                ),
             ('task2', 'task1', 'build'),
             ('task2', 'task3', 'test'),
             self.make_task('task3', optimized=False),
         )
         label_to_taskid = {'task1': 'dep1'}
         sub = get_subgraph(graph, graph.graph.named_links_dict(), label_to_taskid)
         task2 = label_to_taskid['task2']
         task3 = label_to_taskid['task3']
@@ -236,18 +238,19 @@ class TestOptimize(unittest.TestCase):
         self.assertEqual(sub.graph.edges, {(task2, task3, 'test')})
         self.assertEqual(sub.tasks[task2].task_id, task2)
         self.assertEqual(sorted(sub.tasks[task2].task['dependencies']), sorted([task3, 'dep1']))
         self.assertEqual(sub.tasks[task2].task['payload'], 'http://dep1/' + task3)
         self.assertEqual(sub.tasks[task3].task_id, task3)
 
     def test_optimize(self):
         "optimize_task_graph annotates and extracts the subgraph from a simple graph"
-        OptimizingTask.optimize = \
-            lambda self: (True, 'dep1') if self.label == 'task1' else (False, None)
+        self.make_kind(
+            lambda task, deps: (True, 'dep1') if task.label == 'task1' else (False, None)
+            )
         input = self.make_graph(
             self.make_task('task1'),
             self.make_task('task2'),
             self.make_task('task3'),
             ('task2', 'task1', 'build'),
             ('task2', 'task3', 'image'),
         )
         opt, label_to_taskid = optimize_task_graph(input, set())

--- a/taskcluster/taskgraph/test/test_target_tasks.py
+++ b/taskcluster/taskgraph/test/test_target_tasks.py
@@ -4,18 +4,17 @@
 
 from __future__ import absolute_import, print_function, unicode_literals
 
 import unittest
 
 from .. import target_tasks
 from .. import try_option_syntax
 from ..graph import Graph
-from ..taskgraph import TaskGraph
-from .util import TestTask
+from ..types import Task, TaskGraph
 from mozunit import main
 
 
 class FakeTryOptionSyntax(object):
 
     def __init__(self, message, task_graph):
         pass
 
@@ -28,26 +27,26 @@ class TestTargetTasks(unittest.TestCase)
     def test_from_parameters(self):
         method = target_tasks.get_method('from_parameters')
         self.assertEqual(method(None, {'target_tasks': ['a', 'b']}),
                          ['a', 'b'])
 
     def test_all_builds_and_tests(self):
         method = target_tasks.get_method('all_builds_and_tests')
         graph = TaskGraph(tasks={
-            'a': TestTask(kind='legacy', label='a'),
-            'b': TestTask(kind='legacy', label='b'),
-            'boring': TestTask(kind='docker', label='boring'),
+            'a': Task(kind=None, label='a', attributes={'kind': 'legacy'}),
+            'b': Task(kind=None, label='b', attributes={'kind': 'legacy'}),
+            'boring': Task(kind=None, label='boring', attributes={'kind': 'docker-image'}),
         }, graph=Graph(nodes={'a', 'b', 'boring'}, edges=set()))
         self.assertEqual(sorted(method(graph, {})), sorted(['a', 'b']))
 
     def test_try_option_syntax(self):
         tasks = {
-            'a': TestTask(kind=None, label='a'),
-            'b': TestTask(kind=None, label='b', attributes={'at-at': 'yep'}),
+            'a': Task(kind=None, label='a'),
+            'b': Task(kind=None, label='b', attributes={'at-at': 'yep'}),
         }
         graph = Graph(nodes=set('ab'), edges=set())
         tg = TaskGraph(tasks, graph)
         params = {'message': 'try me'}
 
         orig_TryOptionSyntax = try_option_syntax.TryOptionSyntax
         try:
             try_option_syntax.TryOptionSyntax = FakeTryOptionSyntax

--- a/taskcluster/taskgraph/test/test_try_option_syntax.py
+++ b/taskcluster/taskgraph/test/test_try_option_syntax.py
@@ -3,33 +3,32 @@
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
 from __future__ import absolute_import, print_function, unicode_literals
 
 import unittest
 
 from ..try_option_syntax import TryOptionSyntax
 from ..graph import Graph
-from ..taskgraph import TaskGraph
-from .util import TestTask
+from ..types import TaskGraph, Task
 from mozunit import main
 
 # an empty graph, for things that don't look at it
 empty_graph = TaskGraph({}, Graph(set(), set()))
 
 
 def unittest_task(n, tp):
-    return (n, TestTask('test', n, {
+    return (n, Task('test', n, {
         'unittest_try_name': n,
         'test_platform': tp,
     }))
 
 
 def talos_task(n, tp):
-    return (n, TestTask('test', n, {
+    return (n, Task('test', n, {
         'talos_try_name': n,
         'test_platform': tp,
     }))
 
 tasks = {k: v for k, v in [
     unittest_task('mochitest-browser-chrome', 'linux'),
     unittest_task('mochitest-browser-chrome-e10s', 'linux64'),
     unittest_task('mochitest-chrome', 'linux'),
@@ -254,18 +253,18 @@ class TestTryOptionSyntax(unittest.TestC
     def test_t_single(self):
         "-t mochitest-webgl sets talos=[mochitest-webgl]"
         tos = TryOptionSyntax('try: -t mochitest-webgl', graph_with_jobs)
         self.assertEqual(sorted(tos.talos), sorted([{'test': 'mochitest-webgl'}]))
 
     # -t shares an implementation with -u, so it's not tested heavily
 
     def test_trigger_tests(self):
-        "--rebuild 10 sets trigger_tests"
-        tos = TryOptionSyntax('try: --rebuild 10', empty_graph)
+        "--trigger-tests 10 sets trigger_tests"
+        tos = TryOptionSyntax('try: --trigger-tests 10', empty_graph)
         self.assertEqual(tos.trigger_tests, 10)
 
     def test_interactive(self):
         "--interactive sets interactive"
         tos = TryOptionSyntax('try: --interactive', empty_graph)
         self.assertEqual(tos.interactive, True)
 
 if __name__ == '__main__':

deleted file mode 100644
--- a/taskcluster/taskgraph/test/util.py
+++ /dev/null
@@ -1,24 +0,0 @@
-# This Source Code Form is subject to the terms of the Mozilla Public
-# License, v. 2.0. If a copy of the MPL was not distributed with this
-# file, You can obtain one at http://mozilla.org/MPL/2.0/.
-
-from __future__ import absolute_import, print_function, unicode_literals
-
-from ..kind import base
-
-
-class TestTask(base.Task):
-
-    def __init__(self, kind=None, label=None, attributes=None, task=None):
-        super(TestTask, self).__init__(
-                kind or 'test',
-                label or 'test-label',
-                attributes or {},
-                task or {})
-
-    @classmethod
-    def load_tasks(cls, kind, path, config, parameters):
-        return []
-
-    def get_dependencies(self, taskgraph):
-        return []

rename from taskcluster/taskgraph/taskgraph.py
rename to taskcluster/taskgraph/types.py
--- a/taskcluster/taskgraph/taskgraph.py
+++ b/taskcluster/taskgraph/types.py
@@ -1,15 +1,51 @@
 # This Source Code Form is subject to the terms of the Mozilla Public
 # License, v. 2.0. If a copy of the MPL was not distributed with this
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
 from __future__ import absolute_import, print_function, unicode_literals
 
 
+class Task(object):
+    """
+    Representation of a task in a TaskGraph.
+
+    Each has, at creation:
+
+    - kind: Kind instance that created this task
+    - label; the label for this task
+    - attributes: a dictionary of attributes for this task (used for filtering)
+    - task: the task definition (JSON-able dictionary)
+    - extra: extra kind-specific metadata
+
+    And later, as the task-graph processing proceeds:
+
+    - optimization_key -- key for finding equivalent tasks in the TC index
+    - task_id -- TC taskId under which this task will be created
+    """
+
+    def __init__(self, kind, label, attributes=None, task=None, **extra):
+        self.kind = kind
+        self.label = label
+        self.attributes = attributes or {}
+        self.task = task or {}
+        self.extra = extra
+
+        self.task_id = None
+
+        if not (all(isinstance(x, basestring) for x in self.attributes.iterkeys()) and
+                all(isinstance(x, basestring) for x in self.attributes.itervalues())):
+            raise TypeError("attribute names and values must be strings")
+
+    def __str__(self):
+        return "{} ({})".format(self.task_id or self.label,
+                                self.task['metadata']['description'].strip())
+
+
 class TaskGraph(object):
     """
     Representation of a task graph.
 
     A task graph is a combination of a Graph and a dictionary of tasks indexed
     by label.  TaskGraph instances should be treated as immutable.
     """