diff --git a/examples/src/wire/org/pantsbuild/example/element/BUILD b/examples/src/wire/org/pantsbuild/example/element/BUILD
index 916a7f943a1..0afde70a649 100644
--- a/examples/src/wire/org/pantsbuild/example/element/BUILD
+++ b/examples/src/wire/org/pantsbuild/example/element/BUILD
@@ -3,10 +3,12 @@
 
 java_wire_library(
   sources=[
-    'elements.proto', # Order matters here.
+    # NB: Order matters for these two paths, so we set `ordered_sources=True` below.
+    'elements.proto',
     'compound.proto',
   ],
   dependencies=[
     'examples/src/wire/org/pantsbuild/example/temperature',
   ],
+  ordered_sources=True,
 )
diff --git a/src/python/pants/backend/codegen/wire/java/java_wire_library.py b/src/python/pants/backend/codegen/wire/java/java_wire_library.py
index d5cbd9b3fbe..93391af49c4 100644
--- a/src/python/pants/backend/codegen/wire/java/java_wire_library.py
+++ b/src/python/pants/backend/codegen/wire/java/java_wire_library.py
@@ -32,6 +32,7 @@ def __init__(self,
                registry_class=None,
                enum_options=None,
                no_options=None,
+               ordered_sources=None,
                **kwargs):
     """
     :param string service_writer: the name of the class to pass as the --service_writer option to
@@ -43,6 +44,9 @@ def __init__(self,
     doubt, specify com.squareup.wire.SimpleServiceWriter
     :param list enum_options: list of enums to pass to as the --enum-enum_options option, # optional
     :param boolean no_options: boolean that determines if --no_options flag is passed
+    :param boolean ordered_sources: boolean that declares whether the sources argument represents
+      literal ordered sources to be passed directly to the compiler. If false, no ordering is
+      guaranteed for the sources passed to an individual compiler invoke.
     """
 
     if not service_writer and service_writer_options:
@@ -59,6 +63,7 @@ def __init__(self,
       'registry_class': PrimitiveField(registry_class or None),
       'enum_options': PrimitiveField(enum_options or []),
       'no_options': PrimitiveField(no_options or False),
+      'ordered_sources': PrimitiveField(ordered_sources or False),
     })
 
     super(JavaWireLibrary, self).__init__(payload=payload, **kwargs)
diff --git a/src/python/pants/backend/codegen/wire/java/wire_gen.py b/src/python/pants/backend/codegen/wire/java/wire_gen.py
index 84b378d9136..acfb8d58f11 100644
--- a/src/python/pants/backend/codegen/wire/java/wire_gen.py
+++ b/src/python/pants/backend/codegen/wire/java/wire_gen.py
@@ -13,10 +13,12 @@
 from pants.backend.jvm.targets.java_library import JavaLibrary
 from pants.backend.jvm.tasks.nailgun_task import NailgunTaskBase
 from pants.base.build_environment import get_buildroot
-from pants.base.exceptions import TaskError
+from pants.base.exceptions import TargetDefinitionException, TaskError
 from pants.base.workunit import WorkUnitLabel
 from pants.java.jar.jar_dependency import JarDependency
+from pants.source.filespec import globs_matches
 from pants.task.simple_codegen_task import SimpleCodegenTask
+from pants.util.dirutil import fast_relpath
 
 
 logger = logging.getLogger(__name__)
@@ -61,24 +63,47 @@ def synthetic_target_extra_dependencies(self, target, target_workdir):
     wire_runtime_deps_spec = self.get_options().javadeps
     return self.resolve_deps([wire_runtime_deps_spec])
 
-  def format_args_for_target(self, target, target_workdir):
-    """Calculate the arguments to pass to the command line for a single target."""
-    sources = OrderedSet(target.sources_relative_to_buildroot())
-
+  def _compute_sources(self, target):
     relative_sources = OrderedSet()
-    source_roots = set()
-    for source in sources:
+    source_roots = OrderedSet()
+
+    def capture_and_relativize_to_source_root(source):
       source_root = self.context.source_roots.find_by_path(source)
       if not source_root:
         source_root = self.context.source_roots.find(target)
       source_roots.add(source_root.path)
-      relative_source = os.path.relpath(source, source_root.path)
-      relative_sources.add(relative_source)
+      return fast_relpath(source, source_root.path)
+
+    if target.payload.get_field_value('ordered_sources'):
+      # Re-match the filespecs against the sources in order to apply them in the literal order
+      # they were specified in.
+      filespec = target.globs_relative_to_buildroot()
+      excludes = filespec.get('excludes', [])
+      for filespec in filespec.get('globs', []):
+        sources = [s for s in target.sources_relative_to_buildroot()
+                   if globs_matches([s], [filespec], excludes)]
+        if len(sources) != 1:
+          raise TargetDefinitionException(
+              target,
+              'With `ordered_sources=True`, expected one match for each file literal, '
+              'but got: {} for literal `{}`.'.format(sources, filespec)
+            )
+        relative_sources.add(capture_and_relativize_to_source_root(sources[0]))
+    else:
+      # Otherwise, use the default (unspecified) snapshot ordering.
+      for source in target.sources_relative_to_buildroot():
+        relative_sources.add(capture_and_relativize_to_source_root(source))
+    return relative_sources, source_roots
+
+  def format_args_for_target(self, target, target_workdir):
+    """Calculate the arguments to pass to the command line for a single target."""
 
     args = ['--java_out={0}'.format(target_workdir)]
 
     # Add all params in payload to args
 
+    relative_sources, source_roots = self._compute_sources(target)
+
     if target.payload.get_field_value('no_options'):
       args.append('--no_options')
 
diff --git a/src/python/pants/backend/graph_info/tasks/cloc.py b/src/python/pants/backend/graph_info/tasks/cloc.py
index 6019ef33792..e1903d9dfd5 100644
--- a/src/python/pants/backend/graph_info/tasks/cloc.py
+++ b/src/python/pants/backend/graph_info/tasks/cloc.py
@@ -40,7 +40,7 @@ def console_output(self, targets):
     input_snapshots = tuple(
       target.sources_snapshot(scheduler=self.context._scheduler) for target in targets
     )
-    input_files = {f.path for snapshot in input_snapshots for f in snapshot.files}
+    input_files = {f for snapshot in input_snapshots for f in snapshot.files}
 
     # TODO: Work out a nice library-like utility for writing an argfile, as this will be common.
     with temporary_dir() as tmpdir:
diff --git a/src/python/pants/backend/jvm/tasks/jvm_compile/javac/javac_compile.py b/src/python/pants/backend/jvm/tasks/jvm_compile/javac/javac_compile.py
index f9f783947ab..dee55bc492f 100644
--- a/src/python/pants/backend/jvm/tasks/jvm_compile/javac/javac_compile.py
+++ b/src/python/pants/backend/jvm/tasks/jvm_compile/javac/javac_compile.py
@@ -209,8 +209,8 @@ def _execute_hermetic_compile(self, cmd, ctx):
       # Assume no extra .class files to grab. We'll fix up that case soon.
       # Drop the source_root from the file path.
       # Assumes `-d .` has been put in the command.
-      os.path.relpath(f.path.replace('.java', '.class'), ctx.target.target_base)
-      for f in input_snapshot.files if f.path.endswith('.java')
+      os.path.relpath(f.replace('.java', '.class'), ctx.target.target_base)
+      for f in input_snapshot.files if f.endswith('.java')
     )
     exec_process_request = ExecuteProcessRequest(
       argv=tuple(cmd),
diff --git a/src/python/pants/backend/jvm/tasks/jvm_compile/rsc/rsc_compile.py b/src/python/pants/backend/jvm/tasks/jvm_compile/rsc/rsc_compile.py
index 6a5675be561..4266f4f5e02 100644
--- a/src/python/pants/backend/jvm/tasks/jvm_compile/rsc/rsc_compile.py
+++ b/src/python/pants/backend/jvm/tasks/jvm_compile/rsc/rsc_compile.py
@@ -33,8 +33,7 @@
 from pants.java.jar.jar_dependency import JarDependency
 from pants.reporting.reporting_utils import items_to_report_element
 from pants.util.contextutil import Timer
-from pants.util.dirutil import (fast_relpath, fast_relpath_optional, maybe_read_file,
-                                safe_file_dump, safe_mkdir)
+from pants.util.dirutil import fast_relpath, fast_relpath_optional, safe_mkdir
 from pants.util.memo import memoized_property
 
 
@@ -60,22 +59,6 @@ def stdout_contents(wu):
     return f.read().rstrip()
 
 
-def write_digest(output_dir, digest):
-  safe_file_dump(
-    '{}.digest'.format(output_dir),
-    mode='w',
-    payload='{}:{}'.format(digest.fingerprint, digest.serialized_bytes_length))
-
-
-def load_digest(output_dir):
-  read_file = maybe_read_file('{}.digest'.format(output_dir), binary_mode=False)
-  if read_file:
-    fingerprint, length = read_file.split(':')
-    return Digest(fingerprint, int(length))
-  else:
-    return None
-
-
 def _create_desandboxify_fn(possible_path_patterns):
   # Takes a collection of possible canonical prefixes, and returns a function that
   # if it finds a matching prefix, strips the path prior to the prefix and returns it
@@ -132,7 +115,7 @@ def __init__(self, *args, **kwargs):
 
   @classmethod
   def implementation_version(cls):
-    return super(RscCompile, cls).implementation_version() + [('RscCompile', 171)]
+    return super(RscCompile, cls).implementation_version() + [('RscCompile', 172)]
 
   @classmethod
   def register_options(cls, register):
@@ -218,7 +201,7 @@ def pathglob_for(filename):
     def to_classpath_entries(paths, scheduler):
       # list of path ->
       # list of (path, optional<digest>) ->
-      path_and_digests = [(p, load_digest(os.path.dirname(p))) for p in paths]
+      path_and_digests = [(p, Digest.load(os.path.dirname(p))) for p in paths]
       # partition: list of path, list of tuples
       paths_without_digests = [p for (p, d) in path_and_digests if not d]
       if paths_without_digests:
@@ -825,7 +808,7 @@ def _runtool_hermetic(self, main, tool_name, args, distribution, tgt=None, input
       raise TaskError(res.stderr)
 
     if output_dir:
-      write_digest(output_dir, res.output_directory_digest)
+      res.output_directory_digest.dump(output_dir)
       self.context._scheduler.materialize_directories((
         DirectoryToMaterialize(
           # NB the first element here is the root to materialize into, not the dir to snapshot
diff --git a/src/python/pants/backend/python/rules/python_test_runner.py b/src/python/pants/backend/python/rules/python_test_runner.py
index 70c6457f77c..26ac23c39e5 100644
--- a/src/python/pants/backend/python/rules/python_test_runner.py
+++ b/src/python/pants/backend/python/rules/python_test_runner.py
@@ -64,7 +64,7 @@ def run_python_test(transitive_hydrated_target, pytest):
   # pex27, where it should be hermetically provided in some way.
   output_pytest_requirements_pex_filename = 'pytest-with-requirements.pex'
   requirements_pex_argv = [
-    './{}'.format(pex_snapshot.files[0].path),
+    './{}'.format(pex_snapshot.files[0]),
     '--python', python_binary,
     '-e', 'pytest:main',
     '-o', output_pytest_requirements_pex_filename,
diff --git a/src/python/pants/engine/build_files.py b/src/python/pants/engine/build_files.py
index ee65bd92119..84430073f35 100644
--- a/src/python/pants/engine/build_files.py
+++ b/src/python/pants/engine/build_files.py
@@ -219,7 +219,7 @@ def addresses_from_address_families(address_mapper, specs):
   """
   # Capture a Snapshot covering all paths for these Specs, then group by directory.
   snapshot = yield Get(Snapshot, PathGlobs, _spec_to_globs(address_mapper, specs))
-  dirnames = {dirname(f.stat.path) for f in snapshot.files}
+  dirnames = {dirname(f) for f in snapshot.files}
   address_families = yield [Get(AddressFamily, Dir(d)) for d in dirnames]
   address_family_by_directory = {af.namespace: af for af in address_families}
 
diff --git a/src/python/pants/engine/fs.py b/src/python/pants/engine/fs.py
index c202bb3d3f3..8d009f01c06 100644
--- a/src/python/pants/engine/fs.py
+++ b/src/python/pants/engine/fs.py
@@ -4,14 +4,16 @@
 
 from __future__ import absolute_import, division, print_function, unicode_literals
 
+import os
+
 from future.utils import binary_type, text_type
 
-from pants.base.project_tree import Dir, File
 from pants.engine.objects import Collection
 from pants.engine.rules import RootRule
 from pants.option.custom_types import GlobExpansionConjunction
 from pants.option.global_options import GlobMatchErrorBehavior
-from pants.util.objects import datatype
+from pants.util.dirutil import maybe_read_file, safe_delete, safe_file_dump
+from pants.util.objects import Exactly, datatype
 
 
 class FileContent(datatype([('path', text_type), ('content', binary_type)])):
@@ -62,10 +64,6 @@ def __new__(cls, include, exclude=(), glob_match_error_behavior=None, conjunctio
       conjunction=GlobExpansionConjunction.create(conjunction, none_is_default=True))
 
 
-class PathGlobsAndRoot(datatype([('path_globs', PathGlobs), ('root', text_type)])):
-  pass
-
-
 class Digest(datatype([('fingerprint', text_type), ('serialized_bytes_length', int)])):
   """A Digest is a content-digest fingerprint, and a length of underlying content.
 
@@ -84,6 +82,33 @@ class Digest(datatype([('fingerprint', text_type), ('serialized_bytes_length', i
   https://github.com/pantsbuild/pants/issues/5802
   """
 
+  @classmethod
+  def _path(cls, directory):
+    return '{}.digest'.format(directory.rstrip(os.sep))
+
+  @classmethod
+  def clear(cls, directory):
+    """Clear any existing Digest file adjacent to the given directory."""
+    safe_delete(cls._path(directory))
+
+  @classmethod
+  def load(cls, directory):
+    """Load a Digest from a `.digest` file adjacent to the given directory.
+
+    :return: A Digest, or None if the Digest did not exist.
+    """
+    read_file = maybe_read_file(cls._path(directory), binary_mode=False)
+    if read_file:
+      fingerprint, length = read_file.split(':')
+      return Digest(fingerprint, int(length))
+    else:
+      return None
+
+  def dump(self, directory):
+    """Dump this Digest object adjacent to the given directory."""
+    payload = '{}:{}'.format(self.fingerprint, self.serialized_bytes_length)
+    safe_file_dump(self._path(directory), payload=payload, mode='w')
+
   def __repr__(self):
     return '''Digest(fingerprint={}, serialized_bytes_length={})'''.format(
       self.fingerprint,
@@ -94,8 +119,25 @@ def __str__(self):
     return repr(self)
 
 
-class Snapshot(datatype([('directory_digest', Digest), ('path_stats', tuple)])):
-  """A Snapshot is a collection of Files and Dirs fingerprinted by their names/content.
+class PathGlobsAndRoot(datatype([
+    ('path_globs', PathGlobs),
+    ('root', text_type),
+    ('digest_hint', Exactly(Digest, type(None))),
+])):
+  """A set of PathGlobs to capture relative to some root (which may exist outside of the buildroot).
+
+  If the `digest_hint` is set, it must be the Digest that we would expect to get if we were to
+  expand and Digest the globs. The hint is an optimization that allows for bypassing filesystem
+  operations in cases where the expected Digest is known, and the content for the Digest is already
+  stored.
+  """
+
+  def __new__(cls, path_globs, root, digest_hint=None):
+    return super(PathGlobsAndRoot, cls).__new__(cls, path_globs, root, digest_hint)
+
+
+class Snapshot(datatype([('directory_digest', Digest), ('files', tuple), ('dirs', tuple)])):
+  """A Snapshot is a collection of file paths and dir paths fingerprinted by their names/content.
 
   Snapshots are used to make it easier to isolate process execution by fixing the contents
   of the files being operated on and easing their movement to and from isolated execution
@@ -106,22 +148,6 @@ class Snapshot(datatype([('directory_digest', Digest), ('path_stats', tuple)])):
   def is_empty(self):
     return self == EMPTY_SNAPSHOT
 
-  @property
-  def dirs(self):
-    return [p for p in self.path_stats if type(p.stat) == Dir]
-
-  @property
-  def dir_stats(self):
-    return [p.stat for p in self.dirs]
-
-  @property
-  def files(self):
-    return [p for p in self.path_stats if type(p.stat) == File]
-
-  @property
-  def file_stats(self):
-    return [p.stat for p in self.files]
-
 
 class MergedDirectories(datatype([('directories', tuple)])):
   pass
@@ -150,7 +176,8 @@ class UrlToFetch(datatype([('url', text_type), ('digest', Digest)])):
 
 EMPTY_SNAPSHOT = Snapshot(
   directory_digest=EMPTY_DIRECTORY_DIGEST,
-  path_stats=(),
+  files=(),
+  dirs=()
 )
 
 
diff --git a/src/python/pants/engine/native.py b/src/python/pants/engine/native.py
index 5026b055382..de32a5ab155 100644
--- a/src/python/pants/engine/native.py
+++ b/src/python/pants/engine/native.py
@@ -690,10 +690,6 @@ def new_scheduler(self,
                     construct_snapshot,
                     construct_file_content,
                     construct_files_content,
-                    construct_path_stat,
-                    construct_dir,
-                    construct_file,
-                    construct_link,
                     construct_process_result,
                     constraint_address,
                     constraint_path_globs,
@@ -722,10 +718,6 @@ def tc(constraint):
         func(construct_snapshot),
         func(construct_file_content),
         func(construct_files_content),
-        func(construct_path_stat),
-        func(construct_dir),
-        func(construct_file),
-        func(construct_link),
         func(construct_process_result),
         # TypeConstraints.
         tc(constraint_address),
diff --git a/src/python/pants/engine/scheduler.py b/src/python/pants/engine/scheduler.py
index fa4688c64ab..5727c22b135 100644
--- a/src/python/pants/engine/scheduler.py
+++ b/src/python/pants/engine/scheduler.py
@@ -14,8 +14,7 @@
 from pants.base.project_tree import Dir, File, Link
 from pants.build_graph.address import Address
 from pants.engine.fs import (Digest, DirectoryToMaterialize, FileContent, FilesContent,
-                             MergedDirectories, Path, PathGlobs, PathGlobsAndRoot, Snapshot,
-                             UrlToFetch)
+                             MergedDirectories, PathGlobs, PathGlobsAndRoot, Snapshot, UrlToFetch)
 from pants.engine.isolated_process import ExecuteProcessRequest, FallibleExecuteProcessResult
 from pants.engine.native import Function, TypeConstraint, TypeId
 from pants.engine.nodes import Return, Throw
@@ -101,10 +100,6 @@ def __init__(
       construct_snapshot=Snapshot,
       construct_file_content=FileContent,
       construct_files_content=FilesContent,
-      construct_path_stat=Path,
-      construct_dir=Dir,
-      construct_file=File,
-      construct_link=Link,
       construct_process_result=FallibleExecuteProcessResult,
       constraint_address=constraint_for(Address),
       constraint_path_globs=constraint_for(PathGlobs),
diff --git a/src/python/pants/source/filespec.py b/src/python/pants/source/filespec.py
index c36d37176a4..77e967043f5 100644
--- a/src/python/pants/source/filespec.py
+++ b/src/python/pants/source/filespec.py
@@ -9,6 +9,9 @@
 
 def glob_to_regex(pattern):
   """Given a glob pattern, return an equivalent regex expression.
+
+  TODO: Replace with implementation in `fs.rs`. See https://github.com/pantsbuild/pants/issues/6795.
+
   :param string glob: The glob pattern. "**" matches 0 or more dirs recursively.
                       "*" only matches patterns in a single dir.
   :returns: A regex string that matches same paths as the input glob does.
diff --git a/src/python/pants/source/wrapped_globs.py b/src/python/pants/source/wrapped_globs.py
index 2e68472e748..1deab4c9dda 100644
--- a/src/python/pants/source/wrapped_globs.py
+++ b/src/python/pants/source/wrapped_globs.py
@@ -99,8 +99,10 @@ def files(self):
 
   @memoized_property
   def files_relative_to_buildroot(self):
-    fds = self._snapshot.path_stats if self._include_dirs else self._snapshot.files
-    return tuple(fd.path for fd in fds)
+    res = self._snapshot.files
+    if self._include_dirs:
+      res += self._snapshot.dirs
+    return res
 
   @property
   def files_hash(self):
diff --git a/src/python/pants/task/simple_codegen_task.py b/src/python/pants/task/simple_codegen_task.py
index 276c50fe370..162b8dd4d2e 100644
--- a/src/python/pants/task/simple_codegen_task.py
+++ b/src/python/pants/task/simple_codegen_task.py
@@ -18,11 +18,11 @@
 from pants.base.workunit import WorkUnitLabel
 from pants.build_graph.address import Address
 from pants.build_graph.address_lookup_error import AddressLookupError
-from pants.engine.fs import PathGlobs, PathGlobsAndRoot
+from pants.engine.fs import Digest, PathGlobs, PathGlobsAndRoot
 from pants.source.wrapped_globs import EagerFilesetWithSpec, FilesetRelPathWrapper
 from pants.task.task import Task
 from pants.util.collections_abc_backport import OrderedDict
-from pants.util.dirutil import safe_delete
+from pants.util.dirutil import fast_relpath, safe_delete
 
 
 logger = logging.getLogger(__name__)
@@ -113,6 +113,10 @@ def synthetic_target_extra_dependencies(self, target, target_workdir):
     """
     return []
 
+  @classmethod
+  def implementation_version(cls):
+    return super(SimpleCodegenTask, cls).implementation_version() + [('SimpleCodegenTask', 2)]
+
   def synthetic_target_extra_exports(self, target, target_workdir):
     """Gets any extra exports generated synthetic targets should have.
 
@@ -206,7 +210,7 @@ def _do_validate_sources_present(self, target):
 
   def _get_synthetic_address(self, target, target_workdir):
     synthetic_name = target.id
-    sources_rel_path = os.path.relpath(target_workdir, get_buildroot())
+    sources_rel_path = fast_relpath(target_workdir, get_buildroot())
     synthetic_address = Address(sources_rel_path, synthetic_name)
     return synthetic_address
 
@@ -230,32 +234,26 @@ def execute(self):
       with self.context.new_workunit(name='execute', labels=[WorkUnitLabel.MULTITOOL]):
         vts_to_sources = OrderedDict()
         for vt in invalidation_check.all_vts:
-          synthetic_target_dir = self.synthetic_target_dir(vt.target, vt.results_dir)
 
-          key = (vt, synthetic_target_dir)
-          vts_to_sources[key] = None
+          vts_to_sources[vt] = None
 
           # Build the target and handle duplicate sources.
           if not vt.valid:
             if self._do_validate_sources_present(vt.target):
-              self.execute_codegen(vt.target, vt.results_dir)
-              sources = self._capture_sources((key,))[0]
+              self.execute_codegen(vt.target, vt.current_results_dir)
+              sources = self._capture_sources((vt,))[0]
               # _handle_duplicate_sources may delete files from the filesystem, so we need to
               # re-capture the sources.
-              if not self._handle_duplicate_sources(vt.target, vt.results_dir, sources):
-                vts_to_sources[key] = sources
+              if not self._handle_duplicate_sources(vt, sources):
+                vts_to_sources[vt] = sources
             vt.update()
 
         vts_to_capture = tuple(key for key, sources in vts_to_sources.items() if sources is None)
         filesets = self._capture_sources(vts_to_capture)
         for key, fileset in zip(vts_to_capture, filesets):
           vts_to_sources[key] = fileset
-        for (vt, synthetic_target_dir), fileset in vts_to_sources.items():
-          self._inject_synthetic_target(
-            vt.target,
-            synthetic_target_dir,
-            fileset,
-          )
+        for vt, fileset in vts_to_sources.items():
+          self._inject_synthetic_target(vt, fileset)
         self._mark_transitive_invalidation_hashes_dirty(
           vt.target.address for vt in invalidation_check.all_vts
         )
@@ -280,17 +278,23 @@ def synthetic_target_dir(self, target, target_workdir):
     """
     return target_workdir
 
-  # Accepts tuple of tuples of (target, synthetic_target_dir)
+  # Accepts tuple of VersionedTarget instances.
   # Returns tuple of EagerFilesetWithSpecs in matching order.
-  def _capture_sources(self, targets_and_dirs):
+  def _capture_sources(self, vts):
     to_capture = []
     results_dirs = []
     filespecs = []
 
-    for target, synthetic_target_dir in targets_and_dirs:
+    for vt in vts:
+      target = vt.target
+      # Compute the (optional) subdirectory of the results_dir to generate code to. This
+      # path will end up in the generated FilesetWithSpec and target, and thus needs to be
+      # located below the stable/symlinked `vt.results_dir`.
+      synthetic_target_dir = self.synthetic_target_dir(target, vt.results_dir)
+
       files = self.sources_globs
 
-      results_dir_relpath = os.path.relpath(synthetic_target_dir, get_buildroot())
+      results_dir_relpath = fast_relpath(synthetic_target_dir, get_buildroot())
       buildroot_relative_globs = tuple(os.path.join(results_dir_relpath, file) for file in files)
       buildroot_relative_excludes = tuple(
         os.path.join(results_dir_relpath, file)
@@ -300,6 +304,8 @@ def _capture_sources(self, targets_and_dirs):
         PathGlobsAndRoot(
           PathGlobs(buildroot_relative_globs, buildroot_relative_excludes),
           text_type(get_buildroot()),
+          # The digest is stored adjacent to the hash-versioned `vt.current_results_dir`.
+          Digest.load(vt.current_results_dir),
         )
       )
       results_dirs.append(results_dir_relpath)
@@ -307,33 +313,35 @@ def _capture_sources(self, targets_and_dirs):
 
     snapshots = self.context._scheduler.capture_snapshots(tuple(to_capture))
 
+    for snapshot, vt in zip(snapshots, vts):
+      snapshot.directory_digest.dump(vt.current_results_dir)
+
     return tuple(EagerFilesetWithSpec(
       results_dir_relpath,
       filespec,
       snapshot,
     ) for (results_dir_relpath, filespec, snapshot) in zip(results_dirs, filespecs, snapshots))
 
-  def _inject_synthetic_target(
-    self,
-    target,
-    target_workdir,
-    sources,
-  ):
+  def _inject_synthetic_target(self, vt, sources):
     """Create, inject, and return a synthetic target for the given target and workdir.
 
-    :param target: The target to inject a synthetic target for.
-    :param target_workdir: The work directory containing the generated code for the target.
+    :param vt: A codegen input VersionedTarget to inject a synthetic target for.
+    :param sources: A FilesetWithSpec to inject for the target.
     """
+    target = vt.target
 
+    # NB: For stability, the injected target exposes the stable-symlinked `vt.results_dir`,
+    # rather than the hash-named `vt.current_results_dir`.
+    synthetic_target_dir = self.synthetic_target_dir(target, vt.results_dir)
     synthetic_target_type = self.synthetic_target_type(target)
-    synthetic_extra_dependencies = self.synthetic_target_extra_dependencies(target, target_workdir)
+    synthetic_extra_dependencies = self.synthetic_target_extra_dependencies(target, synthetic_target_dir)
 
     copied_attributes = {}
     for attribute in self._copy_target_attributes:
       copied_attributes[attribute] = getattr(target, attribute)
 
     if self._supports_exports(synthetic_target_type):
-      extra_exports = self.synthetic_target_extra_exports(target, target_workdir)
+      extra_exports = self.synthetic_target_extra_exports(target, synthetic_target_dir)
 
       extra_exports_not_in_extra_dependencies = set(extra_exports).difference(
         set(synthetic_extra_dependencies))
@@ -349,7 +357,7 @@ def _inject_synthetic_target(
       copied_attributes['exports'] = sorted(union)
 
     synthetic_target = self.context.add_new_target(
-      address=self._get_synthetic_address(target, target_workdir),
+      address=self._get_synthetic_address(target, synthetic_target_dir),
       target_type=synthetic_target_type,
       dependencies=synthetic_extra_dependencies,
       sources=sources,
@@ -405,7 +413,7 @@ def execute_codegen(self, target, target_workdir):
     :param target_workdir: A clean directory into which to generate code
     """
 
-  def _handle_duplicate_sources(self, target, target_workdir, sources):
+  def _handle_duplicate_sources(self, vt, sources):
     """Handles duplicate sources generated by the given gen target by either failure or deletion.
 
     This method should be called after all dependencies have been injected into the graph, but
@@ -420,6 +428,8 @@ def _handle_duplicate_sources(self, target, target_workdir, sources):
     default, this behavior is disabled, and duplication in generated sources will raise a
     TaskError. This is controlled by the --allow-dups flag.
     """
+    target = vt.target
+    target_workdir = vt.results_dir
 
     # Walk dependency gentargets and record any sources owned by those targets that are also
     # owned by this target.
@@ -457,6 +467,8 @@ def record_duplicates(dep):
       for duped_source in duped_sources:
         safe_delete(os.path.join(target_workdir, duped_source))
         did_modify = True
+    if did_modify:
+      Digest.clear(vt.current_results_dir)
     return did_modify
 
   class DuplicateSourceError(TaskError):
diff --git a/src/rust/engine/fs/src/snapshot.rs b/src/rust/engine/fs/src/snapshot.rs
index 1badce3abc7..3701fd2278e 100644
--- a/src/rust/engine/fs/src/snapshot.rs
+++ b/src/rust/engine/fs/src/snapshot.rs
@@ -3,7 +3,7 @@
 
 use crate::glob_matching::GlobMatching;
 use crate::pool::ResettablePool;
-use crate::{File, PathGlobs, PathStat, PosixFS, Store};
+use crate::{Dir, File, PathGlobs, PathStat, PosixFS, Store};
 use bazel_protos;
 use boxfuture::{try_future, BoxFuture, Boxable};
 use futures::future::{self, join_all};
@@ -44,27 +44,57 @@ impl Snapshot {
   >(
     store: Store,
     file_digester: &S,
-    path_stats: Vec<PathStat>,
+    mut path_stats: Vec<PathStat>,
   ) -> BoxFuture<Snapshot, String> {
-    let mut sorted_path_stats = path_stats.clone();
-    sorted_path_stats.sort_by(|a, b| a.path().cmp(b.path()));
+    path_stats.sort_by(|a, b| a.path().cmp(b.path()));
 
     // The helper assumes that if a Path has multiple children, it must be a directory.
     // Proactively error if we run into identically named files, because otherwise we will treat
     // them like empty directories.
-    sorted_path_stats.dedup_by(|a, b| a.path() == b.path());
-    if sorted_path_stats.len() != path_stats.len() {
+    let pre_dedupe_len = path_stats.len();
+    path_stats.dedup_by(|a, b| a.path() == b.path());
+    if path_stats.len() != pre_dedupe_len {
       return future::err(format!(
         "Snapshots must be constructed from unique path stats; got duplicates in {:?}",
         path_stats
       ))
       .to_boxed();
     }
-    Snapshot::ingest_directory_from_sorted_path_stats(store, file_digester, &sorted_path_stats)
+    Snapshot::ingest_directory_from_sorted_path_stats(store, file_digester, &path_stats)
       .map(|digest| Snapshot { digest, path_stats })
       .to_boxed()
   }
 
+  pub fn from_digest(store: Store, digest: Digest) -> BoxFuture<Snapshot, String> {
+    store
+      .walk(digest, |_, path_so_far, _, directory| {
+        let mut path_stats = Vec::new();
+        path_stats.extend(directory.get_directories().iter().map(move |dir_node| {
+          let path = path_so_far.join(dir_node.get_name());
+          PathStat::dir(path.clone(), Dir(path))
+        }));
+        path_stats.extend(directory.get_files().iter().map(move |file_node| {
+          let path = path_so_far.join(file_node.get_name());
+          PathStat::file(
+            path.clone(),
+            File {
+              path,
+              is_executable: file_node.is_executable,
+            },
+          )
+        }));
+        future::ok(path_stats).to_boxed()
+      })
+      .map(move |path_stats_per_directory| {
+        let mut path_stats =
+          Iterator::flatten(path_stats_per_directory.into_iter().map(|v| v.into_iter()))
+            .collect::<Vec<_>>();
+        path_stats.sort_by(|l, r| l.path().cmp(&r.path()));
+        Snapshot { digest, path_stats }
+      })
+      .to_boxed()
+  }
+
   pub fn digest_from_path_stats<
     S: StoreFileByDigest<Error> + Sized + Clone,
     Error: fmt::Debug + 'static + Send,
@@ -312,29 +342,44 @@ impl Snapshot {
       .to_boxed()
   }
 
-  pub fn capture_snapshot_from_arbitrary_root<P: AsRef<Path>>(
+  ///
+  /// Capture a Snapshot of a presumed-immutable piece of the filesystem.
+  ///
+  /// Note that we don't use a Graph here, and don't cache any intermediate steps, we just place
+  /// the resultant Snapshot into the store and return it. This is important, because we're reading
+  /// things from arbitrary filepaths which we don't want to cache in the graph, as we don't watch
+  /// them for changes.
+  ///
+  /// If the `digest_hint` is given, first attempt to load the Snapshot using that Digest, and only
+  /// fall back to actually walking the filesystem if we don't have it (either due to garbage
+  /// collection or Digest-oblivious legacy caching).
+  ///
+  pub fn capture_snapshot_from_arbitrary_root<P: AsRef<Path> + Send + 'static>(
     store: Store,
     fs_pool: Arc<ResettablePool>,
     root_path: P,
     path_globs: PathGlobs,
+    digest_hint: Option<Digest>,
   ) -> BoxFuture<Snapshot, String> {
-    // Note that we don't use a Graph here, and don't cache any intermediate steps, we just place
-    // the resultant Snapshot into the store and return it. This is important, because we're reading
-    // things from arbitrary filepaths which we don't want to cache in the graph, as we don't watch
-    // them for changes.
-    // We assume that this Snapshot is of an immutable piece of the filesystem.
-
-    let posix_fs = Arc::new(try_future!(PosixFS::new(root_path, fs_pool, &[])));
-
-    posix_fs
-      .expand(path_globs)
-      .map_err(|err| format!("Error expanding globs: {:?}", err))
-      .and_then(|path_stats| {
-        Snapshot::from_path_stats(
-          store.clone(),
-          &OneOffStoreFileByDigest::new(store, posix_fs),
-          path_stats,
-        )
+    // Attempt to use the digest hint to load a Snapshot without expanding the globs; otherwise,
+    // expand the globs to capture a Snapshot.
+    let store2 = store.clone();
+    future::result(digest_hint.ok_or_else(|| "No digest hint provided.".to_string()))
+      .and_then(move |digest| Snapshot::from_digest(store, digest))
+      .or_else(|_| {
+        let posix_fs = Arc::new(try_future!(PosixFS::new(root_path, fs_pool, &[])));
+
+        posix_fs
+          .expand(path_globs)
+          .map_err(|err| format!("Error expanding globs: {:?}", err))
+          .and_then(|path_stats| {
+            Snapshot::from_path_stats(
+              store2.clone(),
+              &OneOffStoreFileByDigest::new(store2, posix_fs),
+              path_stats,
+            )
+          })
+          .to_boxed()
       })
       .to_boxed()
   }
@@ -507,6 +552,27 @@ mod tests {
     );
   }
 
+  #[test]
+  fn snapshot_from_digest() {
+    let (store, dir, posix_fs, digester) = setup();
+
+    let cats = PathBuf::from("cats");
+    let roland = cats.join("roland");
+    std::fs::create_dir_all(&dir.path().join(cats)).unwrap();
+    make_file(&dir.path().join(&roland), STR.as_bytes(), 0o600);
+
+    let path_stats = expand_all_sorted(posix_fs);
+    let expected_snapshot = Snapshot::from_path_stats(store.clone(), &digester, path_stats.clone())
+      .wait()
+      .unwrap();
+    assert_eq!(
+      expected_snapshot,
+      Snapshot::from_digest(store, expected_snapshot.digest)
+        .wait()
+        .unwrap(),
+    );
+  }
+
   #[test]
   fn snapshot_recursive_directories_including_empty() {
     let (store, dir, posix_fs, digester) = setup();
@@ -535,7 +601,7 @@ mod tests {
           .unwrap(),
           232,
         ),
-        path_stats: unsorted_path_stats,
+        path_stats: sorted_path_stats,
       }
     );
   }
diff --git a/src/rust/engine/fs/src/store.rs b/src/rust/engine/fs/src/store.rs
index 1e1f5a2fbcf..412be087044 100644
--- a/src/rust/engine/fs/src/store.rs
+++ b/src/rust/engine/fs/src/store.rs
@@ -461,51 +461,22 @@ impl Store {
   }
 
   pub fn expand_directory(&self, digest: Digest) -> BoxFuture<HashMap<Digest, EntryType>, String> {
-    let accumulator = Arc::new(Mutex::new(HashMap::new()));
-
-    self
-      .expand_directory_helper(digest, accumulator.clone())
-      .map(|()| {
-        Arc::try_unwrap(accumulator)
-          .expect("Arc should have been unwrappable")
-          .into_inner()
-      })
-      .to_boxed()
-  }
-
-  fn expand_directory_helper(
-    &self,
-    digest: Digest,
-    accumulator: Arc<Mutex<HashMap<Digest, EntryType>>>,
-  ) -> BoxFuture<(), String> {
-    let store = self.clone();
     self
-      .load_directory(digest)
-      .and_then(move |maybe_directory| match maybe_directory {
-        Some(directory) => {
-          {
-            let mut accumulator = accumulator.lock();
-            accumulator.insert(digest, EntryType::Directory);
-            for file in directory.get_files() {
-              accumulator.insert(try_future!(file.get_digest().into()), EntryType::File);
-            }
-          }
-          future::join_all(
-            directory
-              .get_directories()
-              .iter()
-              .map(move |subdir| {
-                store.clone().expand_directory_helper(
-                  try_future!(subdir.get_digest().into()),
-                  accumulator.clone(),
-                )
-              })
-              .collect::<Vec<_>>(),
-          )
-          .map(|_| ())
-          .to_boxed()
+      .walk(digest, |_, _, digest, directory| {
+        let mut digest_types = Vec::new();
+        digest_types.push((digest, EntryType::Directory));
+        for file in directory.get_files() {
+          digest_types.push((try_future!(file.get_digest().into()), EntryType::File));
         }
-        None => future::err(format!("Could not expand unknown directory: {:?}", digest)).to_boxed(),
+        future::ok(digest_types).to_boxed()
+      })
+      .map(|digest_pairs_per_directory| {
+        Iterator::flatten(
+          digest_pairs_per_directory
+            .into_iter()
+            .map(|v| v.into_iter()),
+        )
+        .collect()
       })
       .to_boxed()
   }
@@ -579,78 +550,124 @@ impl Store {
   }
 
   // Returns files sorted by their path.
-  pub fn contents_for_directory(
-    &self,
-    directory: &bazel_protos::remote_execution::Directory,
-  ) -> BoxFuture<Vec<FileContent>, String> {
-    let accumulator = Arc::new(Mutex::new(HashMap::new()));
+  pub fn contents_for_directory(&self, digest: Digest) -> BoxFuture<Vec<FileContent>, String> {
     self
-      .contents_for_directory_helper(directory, PathBuf::new(), accumulator.clone())
-      .map(|()| {
-        let map = Arc::try_unwrap(accumulator).unwrap().into_inner();
-        let mut vec: Vec<FileContent> = map
-          .into_iter()
-          .map(|(path, content)| FileContent { path, content })
-          .collect();
+      .walk(digest, |store, path_so_far, _, directory| {
+        future::join_all(
+          directory
+            .get_files()
+            .iter()
+            .map(move |file_node| {
+              let path = path_so_far.join(file_node.get_name());
+              store
+                .load_file_bytes_with(try_future!(file_node.get_digest().into()), |b| b)
+                .and_then(move |maybe_bytes| {
+                  maybe_bytes
+                    .ok_or_else(|| format!("Couldn't find file contents for {:?}", path))
+                    .map(|content| FileContent { path, content })
+                })
+                .to_boxed()
+            })
+            .collect::<Vec<_>>(),
+        )
+        .to_boxed()
+      })
+      .map(|file_contents_per_directory| {
+        let mut vec = Iterator::flatten(
+          file_contents_per_directory
+            .into_iter()
+            .map(|v| v.into_iter()),
+        )
+        .collect::<Vec<_>>();
         vec.sort_by(|l, r| l.path.cmp(&r.path));
         vec
       })
       .to_boxed()
   }
 
-  // Assumes that all fingerprints it encounters are valid.
-  fn contents_for_directory_helper(
+  ///
+  /// Given the Digest for a Directory, recursively walk the Directory, calling the given function
+  /// with the path so far, and the new Directory.
+  ///
+  /// The recursive walk will proceed concurrently, so if order matters, a caller should sort the
+  /// output after the call.
+  ///
+  pub fn walk<
+    T: Send + 'static,
+    F: Fn(
+        &Store,
+        &PathBuf,
+        Digest,
+        &bazel_protos::remote_execution::Directory,
+      ) -> BoxFuture<T, String>
+      + Send
+      + Sync
+      + 'static,
+  >(
     &self,
-    directory: &bazel_protos::remote_execution::Directory,
+    digest: Digest,
+    f: F,
+  ) -> BoxFuture<Vec<T>, String> {
+    let f = Arc::new(f);
+    let accumulator = Arc::new(Mutex::new(Vec::new()));
+    self
+      .walk_helper(digest, PathBuf::new(), f, accumulator.clone())
+      .map(|()| {
+        Arc::try_unwrap(accumulator)
+          .unwrap_or_else(|_| panic!("walk_helper violated its contract."))
+          .into_inner()
+      })
+      .to_boxed()
+  }
+
+  fn walk_helper<
+    T: Send + 'static,
+    F: Fn(
+        &Store,
+        &PathBuf,
+        Digest,
+        &bazel_protos::remote_execution::Directory,
+      ) -> BoxFuture<T, String>
+      + Send
+      + Sync
+      + 'static,
+  >(
+    &self,
+    digest: Digest,
     path_so_far: PathBuf,
-    contents_wrapped: Arc<Mutex<HashMap<PathBuf, Bytes>>>,
+    f: Arc<F>,
+    accumulator: Arc<Mutex<Vec<T>>>,
   ) -> BoxFuture<(), String> {
-    let contents_wrapped_copy = contents_wrapped.clone();
-    let path_so_far_copy = path_so_far.clone();
-    let store_copy = self.clone();
-    let file_futures = future::join_all(
-      directory
-        .get_files()
-        .iter()
-        .map(move |file_node| {
-          let path = path_so_far_copy.join(file_node.get_name());
-          let contents_wrapped_copy = contents_wrapped_copy.clone();
-          store_copy
-            .load_file_bytes_with(try_future!(file_node.get_digest().into()), |b| b)
-            .and_then(move |maybe_bytes| {
-              maybe_bytes
-                .ok_or_else(|| format!("Couldn't find file contents for {:?}", path))
-                .map(move |bytes| {
-                  let mut contents = contents_wrapped_copy.lock();
-                  contents.insert(path, bytes);
-                })
-            })
-            .to_boxed()
-        })
-        .collect::<Vec<_>>(),
-    );
     let store = self.clone();
-    let dir_futures = future::join_all(
-      directory
-        .get_directories()
-        .iter()
-        .map(move |dir_node| {
-          let digest = try_future!(dir_node.get_digest().into());
-          let path = path_so_far.join(dir_node.get_name());
-          let store = store.clone();
-          let contents_wrapped = contents_wrapped.clone();
-          store
-            .load_directory(digest)
-            .and_then(move |maybe_dir| {
-              maybe_dir
-                .ok_or_else(|| format!("Could not find sub-directory with digest {:?}", digest))
+    self
+      .load_directory(digest)
+      .and_then(move |maybe_directory| match maybe_directory {
+        Some(directory) => {
+          let result_for_directory = f(&store, &path_so_far, digest, &directory);
+          result_for_directory
+            .and_then(move |r| {
+              {
+                let mut accumulator = accumulator.lock();
+                accumulator.push(r);
+              }
+              future::join_all(
+                directory
+                  .get_directories()
+                  .iter()
+                  .map(move |dir_node| {
+                    let subdir_digest = try_future!(dir_node.get_digest().into());
+                    let path = path_so_far.join(dir_node.get_name());
+                    store.walk_helper(subdir_digest, path, f.clone(), accumulator.clone())
+                  })
+                  .collect::<Vec<_>>(),
+              )
+              .map(|_| ())
             })
-            .and_then(move |dir| store.contents_for_directory_helper(&dir, path, contents_wrapped))
             .to_boxed()
-        })
-        .collect::<Vec<_>>(),
-    );
-    file_futures.join(dir_futures).map(|(_, _)| ()).to_boxed()
+        }
+        None => future::err(format!("Could not walk unknown directory: {:?}", digest)).to_boxed(),
+      })
+      .to_boxed()
   }
 }
 
@@ -3501,7 +3518,7 @@ mod tests {
     let store = new_local_store(store_dir.path());
 
     let file_contents = store
-      .contents_for_directory(&TestDirectory::empty().directory())
+      .contents_for_directory(TestDirectory::empty().digest())
       .wait()
       .expect("Getting FileContents");
 
@@ -3535,7 +3552,7 @@ mod tests {
       .expect("Error saving catnip file bytes");
 
     let file_contents = store
-      .contents_for_directory(&recursive_testdir.directory())
+      .contents_for_directory(recursive_testdir.digest())
       .wait()
       .expect("Getting FileContents");
 
diff --git a/src/rust/engine/src/lib.rs b/src/rust/engine/src/lib.rs
index 54a0ed4af0c..84cc83c1fd4 100644
--- a/src/rust/engine/src/lib.rs
+++ b/src/rust/engine/src/lib.rs
@@ -177,10 +177,6 @@ pub extern "C" fn scheduler_create(
   construct_snapshot: Function,
   construct_file_content: Function,
   construct_files_content: Function,
-  construct_path_stat: Function,
-  construct_dir: Function,
-  construct_file: Function,
-  construct_link: Function,
   construct_process_result: Function,
   type_address: TypeConstraint,
   type_path_globs: TypeConstraint,
@@ -224,10 +220,6 @@ pub extern "C" fn scheduler_create(
     construct_snapshot: construct_snapshot,
     construct_file_content: construct_file_content,
     construct_files_content: construct_files_content,
-    construct_path_stat: construct_path_stat,
-    construct_dir: construct_dir,
-    construct_file: construct_file,
-    construct_link: construct_link,
     construct_process_result: construct_process_result,
     address: type_address,
     path_globs: type_path_globs,
@@ -666,15 +658,24 @@ pub extern "C" fn capture_snapshots(
   path_globs_and_root_tuple_wrapper: Handle,
 ) -> PyResult {
   let values = externs::project_multi(&path_globs_and_root_tuple_wrapper.into(), "dependencies");
-  let path_globs_and_roots_result: Result<Vec<(fs::PathGlobs, PathBuf)>, String> = values
+  let path_globs_and_roots_result = values
     .iter()
     .map(|value| {
       let root = PathBuf::from(externs::project_str(&value, "root"));
       let path_globs =
         nodes::Snapshot::lift_path_globs(&externs::project_ignoring_type(&value, "path_globs"));
-      path_globs.map(|path_globs| (path_globs, root))
+      let digest_hint = {
+        let maybe_digest = externs::project_ignoring_type(&value, "digest_hint");
+        // TODO: Extract a singleton Key for None.
+        if maybe_digest == externs::eval("None").unwrap() {
+          None
+        } else {
+          Some(nodes::lift_digest(&maybe_digest)?)
+        }
+      };
+      path_globs.map(|path_globs| (path_globs, root, digest_hint))
     })
-    .collect();
+    .collect::<Result<Vec<_>, _>>();
 
   let path_globs_and_roots = match path_globs_and_roots_result {
     Ok(v) => v,
@@ -689,13 +690,14 @@ pub extern "C" fn capture_snapshots(
     futures::future::join_all(
       path_globs_and_roots
         .into_iter()
-        .map(|(path_globs, root)| {
+        .map(|(path_globs, root, digest_hint)| {
           let core = core.clone();
           fs::Snapshot::capture_snapshot_from_arbitrary_root(
             core.store(),
             core.fs_pool.clone(),
             root,
             path_globs,
+            digest_hint,
           )
           .map(move |snapshot| nodes::Snapshot::store_snapshot(&core, &snapshot))
         })
diff --git a/src/rust/engine/src/nodes.rs b/src/rust/engine/src/nodes.rs
index 3a9587e741e..6220326cf83 100644
--- a/src/rust/engine/src/nodes.rs
+++ b/src/rust/engine/src/nodes.rs
@@ -216,22 +216,11 @@ impl WrappedNode for Select {
               lift_digest(&directory_digest_val).map_err(|str| throw(&str))
             })
             .and_then(move |digest| {
-              let store = context.core.store();
               context
                 .core
                 .store()
-                .load_directory(digest)
+                .contents_for_directory(digest)
                 .map_err(|str| throw(&str))
-                .and_then(move |maybe_directory| {
-                  maybe_directory
-                    .ok_or_else(|| format!("Could not find directory with digest {:?}", digest))
-                    .map_err(|str| throw(&str))
-                })
-                .and_then(move |directory| {
-                  store
-                    .contents_for_directory(&directory)
-                    .map_err(|str| throw(&str))
-                })
                 .map(move |files_content| Snapshot::store_files_content(&context, &files_content))
             })
             .to_boxed()
@@ -540,16 +529,24 @@ impl Snapshot {
   }
 
   pub fn store_snapshot(core: &Arc<Core>, item: &fs::Snapshot) -> Value {
-    let path_stats: Vec<_> = item
-      .path_stats
-      .iter()
-      .map(|ps| Self::store_path_stat(core, ps))
-      .collect();
+    let mut files = Vec::new();
+    let mut dirs = Vec::new();
+    for ps in &item.path_stats {
+      match ps {
+        &PathStat::File { ref path, .. } => {
+          files.push(Self::store_path(path));
+        }
+        &PathStat::Dir { ref path, .. } => {
+          dirs.push(Self::store_path(path));
+        }
+      }
+    }
     externs::unsafe_call(
       &core.types.construct_snapshot,
       &[
         Self::store_directory(core, &item.digest),
-        externs::store_tuple(&path_stats),
+        externs::store_tuple(&files),
+        externs::store_tuple(&dirs),
       ],
     )
   }
@@ -558,28 +555,6 @@ impl Snapshot {
     externs::store_utf8_osstr(item.as_os_str())
   }
 
-  fn store_dir(core: &Arc<Core>, item: &Dir) -> Value {
-    let args = [Self::store_path(item.0.as_path())];
-    externs::unsafe_call(&core.types.construct_dir, &args)
-  }
-
-  fn store_file(core: &Arc<Core>, item: &File) -> Value {
-    let args = [Self::store_path(item.path.as_path())];
-    externs::unsafe_call(&core.types.construct_file, &args)
-  }
-
-  fn store_path_stat(core: &Arc<Core>, item: &PathStat) -> Value {
-    let args = match item {
-      &PathStat::Dir { ref path, ref stat } => {
-        vec![Self::store_path(path), Self::store_dir(core, stat)]
-      }
-      &PathStat::File { ref path, ref stat } => {
-        vec![Self::store_path(path), Self::store_file(core, stat)]
-      }
-    };
-    externs::unsafe_call(&core.types.construct_path_stat, &args)
-  }
-
   fn store_file_content(context: &Context, item: &FileContent) -> Value {
     externs::unsafe_call(
       &context.core.types.construct_file_content,
diff --git a/src/rust/engine/src/types.rs b/src/rust/engine/src/types.rs
index c4333deb737..3b517e7919c 100644
--- a/src/rust/engine/src/types.rs
+++ b/src/rust/engine/src/types.rs
@@ -5,10 +5,6 @@ pub struct Types {
   pub construct_snapshot: Function,
   pub construct_file_content: Function,
   pub construct_files_content: Function,
-  pub construct_path_stat: Function,
-  pub construct_dir: Function,
-  pub construct_file: Function,
-  pub construct_link: Function,
   pub construct_process_result: Function,
   pub address: TypeConstraint,
   pub path_globs: TypeConstraint,
diff --git a/tests/python/pants_test/backend/codegen/antlr/java/test_antlr_java_gen.py b/tests/python/pants_test/backend/codegen/antlr/java/test_antlr_java_gen.py
index c775a60e2a7..d7406a6d71a 100644
--- a/tests/python/pants_test/backend/codegen/antlr/java/test_antlr_java_gen.py
+++ b/tests/python/pants_test/backend/codegen/antlr/java/test_antlr_java_gen.py
@@ -17,9 +17,16 @@
 from pants.base.exceptions import TaskError
 from pants.build_graph.build_file_aliases import BuildFileAliases
 from pants.util.dirutil import safe_mkdtemp
+from pants.util.objects import datatype
 from pants_test.jvm.nailgun_task_test_base import NailgunTaskTestBase
 
 
+class DummyVersionedTarget(datatype(['target', 'results_dir'])):
+  @property
+  def current_results_dir(self):
+    return self.results_dir
+
+
 class AntlrJavaGenTest(NailgunTaskTestBase):
   @classmethod
   def task_type(cls):
@@ -74,11 +81,12 @@ def execute_antlr_test(self, expected_package, target_workdir_fun=None):
     # Do not use task.workdir here, because when we calculating hash for synthetic target
     # we need persistent source paths in terms of relative position to build root.
     target_workdir = target_workdir_fun(self.build_root)
+    vt = DummyVersionedTarget(target, target_workdir)
 
     # Generate code, then create a synthetic target.
     task.execute_codegen(target, target_workdir)
-    sources = task._capture_sources(((target, target_workdir),))[0]
-    syn_target = task._inject_synthetic_target(target, target_workdir, sources)
+    sources = task._capture_sources((vt,))[0]
+    syn_target = task._inject_synthetic_target(vt, sources)
 
     actual_sources = [s for s in Fileset.rglobs('*.java', root=target_workdir)]
     expected_sources = syn_target.sources_relative_to_source_root()
diff --git a/tests/python/pants_test/backend/codegen/protobuf/java/BUILD b/tests/python/pants_test/backend/codegen/protobuf/java/BUILD
index da9c887c158..90451d248da 100644
--- a/tests/python/pants_test/backend/codegen/protobuf/java/BUILD
+++ b/tests/python/pants_test/backend/codegen/protobuf/java/BUILD
@@ -26,4 +26,5 @@ python_tests(
     'tests/python/pants_test:int-test',
   ],
   tags = {'integration'},
+  timeout = 240,
 )
diff --git a/tests/python/pants_test/backend/codegen/protobuf/java/test_protobuf_integration.py b/tests/python/pants_test/backend/codegen/protobuf/java/test_protobuf_integration.py
index 4c1c944cfa7..4075f0ea228 100644
--- a/tests/python/pants_test/backend/codegen/protobuf/java/test_protobuf_integration.py
+++ b/tests/python/pants_test/backend/codegen/protobuf/java/test_protobuf_integration.py
@@ -112,9 +112,7 @@ def find_protoc_blocks(lines):
                 out=pants_run.stderr_data,
                 blocks=block_text))
 
-    biggest_proto = -1
     for block in all_blocks:
-      last_proto = -1
       seen_extracted = False
       for line in block:
         # Make sure import bases appear after the bases for actual sources.
@@ -124,14 +122,3 @@ def find_protoc_blocks(lines):
           else:
             self.assertFalse(seen_extracted,
                              'Local protoc bases must be ordered before imported bases!')
-          continue
-        # Check to make sure, eg, testproto4.proto never precedes testproto2.proto.
-        match = re.search(r'(?P<sequence>\d+)\.proto[\\.]?$', line)
-        if match:
-          number = int(match.group('sequence'))
-          self.assertTrue(number > last_proto, '{proto} succeeded proto #{number}!\n{blocks}'
-                          .format(proto=line, number=last_proto, blocks=block_text))
-          last_proto = number
-      if last_proto > biggest_proto:
-        biggest_proto = last_proto
-    self.assertEqual(biggest_proto, 6, 'Not all protos were seen!\n{}'.format(block_text))
diff --git a/tests/python/pants_test/backend/codegen/wire/java/BUILD b/tests/python/pants_test/backend/codegen/wire/java/BUILD
index 99e8abec771..03c82fe8d84 100644
--- a/tests/python/pants_test/backend/codegen/wire/java/BUILD
+++ b/tests/python/pants_test/backend/codegen/wire/java/BUILD
@@ -6,6 +6,7 @@ python_tests(
   sources = globs('*.py', exclude=[globs('*_integration.py')]),
   dependencies = [
     '3rdparty/python/twitter/commons:twitter.common.collections',
+    '3rdparty/python:parameterized',
     'src/python/pants/backend/codegen/wire/java',
     'src/python/pants/java/jar',
     'src/python/pants/backend/jvm/targets:jvm',
@@ -29,4 +30,5 @@ python_tests(
     'tests/python/pants_test:int-test',
   ],
   tags = {'integration'},
+  timeout = 300,
 )
diff --git a/tests/python/pants_test/backend/codegen/wire/java/test_wire_gen.py b/tests/python/pants_test/backend/codegen/wire/java/test_wire_gen.py
index e05d37b0f4e..b297af56110 100644
--- a/tests/python/pants_test/backend/codegen/wire/java/test_wire_gen.py
+++ b/tests/python/pants_test/backend/codegen/wire/java/test_wire_gen.py
@@ -4,6 +4,8 @@
 
 from __future__ import absolute_import, division, print_function, unicode_literals
 
+from parameterized import parameterized
+
 from pants.backend.codegen.wire.java.java_wire_library import JavaWireLibrary
 from pants.backend.codegen.wire.java.register import build_file_aliases as register_codegen
 from pants.backend.codegen.wire.java.wire_gen import WireGen
@@ -59,28 +61,35 @@ def test_compiler_args_wirev1(self):
       'bar.proto'],
       task.format_args_for_target(wire_targetv1, self.TARGET_WORKDIR))
 
-  def test_compiler_args_all(self):
+  @parameterized.expand([(True,), (False,)])
+  def test_compiler_args_all(self, ordered_sources):
     self._create_fake_wire_tool(version='1.8.0')
     kitchen_sink = self.make_target('src/wire:kitchen-sink', JavaWireLibrary,
                                     sources=['foo.proto', 'bar.proto', 'baz.proto'],
                                     registry_class='org.pantsbuild.Registry',
                                     service_writer='org.pantsbuild.DummyServiceWriter',
                                     no_options=True,
+                                    ordered_sources=ordered_sources,
                                     roots=['root1', 'root2', 'root3'],
                                     enum_options=['enum1', 'enum2', 'enum3'],)
     task = self.create_task(self.context(target_roots=[kitchen_sink]))
-    self.assertEqual([
-      '--java_out={}'.format(self.TARGET_WORKDIR),
-      '--no_options',
-      '--service_writer=org.pantsbuild.DummyServiceWriter',
-      '--registry_class=org.pantsbuild.Registry',
-      '--roots=root1,root2,root3',
-      '--enum_options=enum1,enum2,enum3',
-      '--proto_path={}/src/wire'.format(self.build_root),
-      'foo.proto',
-      'bar.proto',
-      'baz.proto'],
-      task.format_args_for_target(kitchen_sink, self.TARGET_WORKDIR))
+    expected = [
+        '--java_out={}'.format(self.TARGET_WORKDIR),
+        '--no_options',
+        '--service_writer=org.pantsbuild.DummyServiceWriter',
+        '--registry_class=org.pantsbuild.Registry',
+        '--roots=root1,root2,root3',
+        '--enum_options=enum1,enum2,enum3',
+        '--proto_path={}/src/wire'.format(self.build_root),
+        'foo.proto',
+        'bar.proto',
+        'baz.proto',
+      ]
+    actual = task.format_args_for_target(kitchen_sink, self.TARGET_WORKDIR)
+    if not ordered_sources:
+      expected = set(expected)
+      actual = set(actual)
+    self.assertEqual(expected, actual)
 
   def test_compiler_args_proto_paths(self):
     self._create_fake_wire_tool()
diff --git a/tests/python/pants_test/engine/legacy/test_graph.py b/tests/python/pants_test/engine/legacy/test_graph.py
index 959612a9d63..0ca73b013bc 100644
--- a/tests/python/pants_test/engine/legacy/test_graph.py
+++ b/tests/python/pants_test/engine/legacy/test_graph.py
@@ -61,8 +61,9 @@ def test_invalidate_fsnode(self):
     self.assertGreater(invalidated_count, 0)
 
   def test_sources_ordering(self):
-    expected_sources = ['p', 'a', 'n', 't', 's', 'b', 'u', 'i', 'l', 'd']
-    self.create_library('src/example', 'files', 'things', sources=expected_sources)
+    input_sources = ['p', 'a', 'n', 't', 's', 'b', 'u', 'i', 'l', 'd']
+    expected_sources = sorted(input_sources)
+    self.create_library('src/example', 'files', 'things', sources=input_sources)
 
     target = self.target('src/example:things')
     sources = [os.path.basename(s) for s in target.sources_relative_to_buildroot()]
diff --git a/tests/python/pants_test/engine/test_build_files.py b/tests/python/pants_test/engine/test_build_files.py
index 1c5c94850f8..625d3c7ab57 100644
--- a/tests/python/pants_test/engine/test_build_files.py
+++ b/tests/python/pants_test/engine/test_build_files.py
@@ -8,14 +8,13 @@
 import re
 import unittest
 
-from pants.base.project_tree import Dir, File
+from pants.base.project_tree import Dir
 from pants.base.specs import SiblingAddresses, SingleAddress, Specs
 from pants.build_graph.address import Address
 from pants.engine.addressable import addressable, addressable_dict
 from pants.engine.build_files import (ResolvedTypeMismatchError, addresses_from_address_families,
                                       create_graph_rules, parse_address_family)
-from pants.engine.fs import (Digest, FileContent, FilesContent, Path, PathGlobs, Snapshot,
-                             create_fs_rules)
+from pants.engine.fs import Digest, FileContent, FilesContent, PathGlobs, Snapshot, create_fs_rules
 from pants.engine.legacy.structs import TargetAdaptor
 from pants.engine.mapper import AddressFamily, AddressMapper, ResolveError
 from pants.engine.nodes import Return, Throw
@@ -34,7 +33,7 @@ def test_empty(self):
     """Test that parsing an empty BUILD file results in an empty AddressFamily."""
     address_mapper = AddressMapper(JsonParser(TestTable()))
     af = run_rule(parse_address_family, address_mapper, Dir('/dev/null'), {
-        (Snapshot, PathGlobs): lambda _: Snapshot(Digest('abc', 10), (File('/dev/null/BUILD'),)),
+        (Snapshot, PathGlobs): lambda _: Snapshot(Digest('abc', 10), ('/dev/null/BUILD',), ()),
         (FilesContent, Digest): lambda _: FilesContent([FileContent('/dev/null/BUILD', b'')]),
       })
     self.assertEqual(len(af.objects_by_name), 0)
@@ -46,9 +45,7 @@ def _address_mapper(self):
     return AddressMapper(JsonParser(TestTable()))
 
   def _snapshot(self):
-    return Snapshot(
-      Digest('xx', 2),
-      (Path('root/BUILD', File('root/BUILD')),))
+    return Snapshot(Digest('xx', 2), ('root/BUILD',), ())
 
   def _resolve_build_file_addresses(self, specs, address_family, snapshot, address_mapper):
     return run_rule(addresses_from_address_families, address_mapper, specs, {
@@ -59,8 +56,7 @@ def _resolve_build_file_addresses(self, specs, address_family, snapshot, address
   def test_duplicated(self):
     """Test that matching the same Spec twice succeeds."""
     address = SingleAddress('a', 'a')
-    snapshot = Snapshot(Digest('xx', 2),
-                        (Path('a/BUILD', File('a/BUILD')),))
+    snapshot = Snapshot(Digest('xx', 2), ('a/BUILD',), ())
     address_family = AddressFamily('a', {'a': ('a/BUILD', 'this is an object!')})
     specs = Specs([address, address])
 
diff --git a/tests/python/pants_test/engine/test_fs.py b/tests/python/pants_test/engine/test_fs.py
index 4f6c8cfaa8d..ecfdf29afc6 100644
--- a/tests/python/pants_test/engine/test_fs.py
+++ b/tests/python/pants_test/engine/test_fs.py
@@ -61,7 +61,7 @@ def assert_walk_snapshot(self, field, filespecs_or_globs, paths, ignore_patterns
       if prepare:
         prepare(project_tree)
       result = self.execute(scheduler, Snapshot, self.specs(filespecs_or_globs))[0]
-      self.assertEqual(sorted([p.path for p in getattr(result, field)]), sorted(paths))
+      self.assertEqual(sorted(getattr(result, field)), sorted(paths))
 
   def assert_content(self, filespecs_or_globs, expected_content):
     with self.mk_project_tree() as project_tree:
@@ -76,7 +76,7 @@ def assert_digest(self, filespecs_or_globs, expected_files):
       scheduler = self.mk_scheduler(rules=create_fs_rules(), project_tree=project_tree)
       result = self.execute(scheduler, Snapshot, self.specs(filespecs_or_globs))[0]
       # Confirm all expected files were digested.
-      self.assertEqual(set(expected_files), {f.path for f in result.files})
+      self.assertEqual(set(expected_files), set(result.files))
       self.assertTrue(result.directory_digest.fingerprint is not None)
 
   def test_walk_literal(self):
@@ -270,7 +270,7 @@ def test_snapshot_from_outside_buildroot_failure(self):
       self.assertIn("doesnotexist", str(cm.exception))
 
   def assert_snapshot_equals(self, snapshot, files, directory_digest):
-    self.assertEqual([file.path for file in snapshot.files], files)
+    self.assertEqual(list(snapshot.files), files)
     self.assertEqual(snapshot.directory_digest, directory_digest)
 
   def test_merge_zero_directories(self):
diff --git a/tests/python/pants_test/engine/test_isolated_process.py b/tests/python/pants_test/engine/test_isolated_process.py
index 5a47b1412d0..3e62fda72f7 100644
--- a/tests/python/pants_test/engine/test_isolated_process.py
+++ b/tests/python/pants_test/engine/test_isolated_process.py
@@ -53,7 +53,7 @@ def bin_path(self):
     return self.binary_location.bin_path
 
   def argv_from_snapshot(self, snapshot):
-    cat_file_paths = [f.path for f in snapshot.files]
+    cat_file_paths = snapshot.files
 
     option_like_files = [p for p in cat_file_paths if p.startswith('-')]
     if option_like_files:
@@ -138,9 +138,7 @@ def bin_path(self):
     return self.binary_location.bin_path
 
   def argv_from_source_snapshot(self, snapshot):
-    snapshot_file_paths = [f.path for f in snapshot.files]
-
-    return (self.bin_path,) + tuple(snapshot_file_paths)
+    return (self.bin_path,) + snapshot.files
 
 
 class JavacCompileResult(datatype([
diff --git a/tests/python/pants_test/reporting/BUILD b/tests/python/pants_test/reporting/BUILD
index e5614eb33a5..eaaf806cc1b 100644
--- a/tests/python/pants_test/reporting/BUILD
+++ b/tests/python/pants_test/reporting/BUILD
@@ -22,7 +22,7 @@ python_tests(
     'tests/python/pants_test:int-test',
   ],
   tags = {'integration'},
-  timeout = 240,
+  timeout = 600,
 )
 
 python_tests(
diff --git a/tests/python/pants_test/source/test_payload_fields.py b/tests/python/pants_test/source/test_payload_fields.py
index 7bb5ff42694..189bf1f9ff2 100644
--- a/tests/python/pants_test/source/test_payload_fields.py
+++ b/tests/python/pants_test/source/test_payload_fields.py
@@ -6,8 +6,7 @@
 
 from future.utils import text_type
 
-from pants.base.project_tree import File
-from pants.engine.fs import Digest, Path, Snapshot
+from pants.engine.fs import Digest, Snapshot
 from pants.source.payload_fields import SourcesField
 from pants.source.wrapped_globs import Globs, LazyFilesetWithSpec
 from pants_test.test_base import TestBase
@@ -83,10 +82,7 @@ def test_passes_eager_fileset_with_spec_through(self):
     self.assertEqual(['foo/foo/a.txt'], list(sf.relative_to_buildroot()))
 
     digest = '56001a7e48555f156420099a99da60a7a83acc90853046709341bf9f00a6f944'
-    want_snapshot = Snapshot(
-      Digest(text_type(digest), 77),
-      (Path('foo/foo/a.txt', stat=File('foo/foo/a.txt')),)
-    )
+    want_snapshot = Snapshot(Digest(text_type(digest), 77), ('foo/foo/a.txt',), ())
 
     # We explicitly pass a None scheduler because we expect no scheduler lookups to be required
     # in order to get a Snapshot.
diff --git a/tests/python/pants_test/source/test_wrapped_globs.py b/tests/python/pants_test/source/test_wrapped_globs.py
index c4a0a01500e..331f72af843 100644
--- a/tests/python/pants_test/source/test_wrapped_globs.py
+++ b/tests/python/pants_test/source/test_wrapped_globs.py
@@ -266,10 +266,6 @@ def test_source_snapshot(self):
     self.add_to_build_file('package/dir', 'files(name = "target", sources = ["foo"])')
     target = self.target('package/dir:target')
     snapshot = target.sources_snapshot(scheduler=self.scheduler)
-    snapshot_paths = tuple(file.path for file in snapshot.path_stats)
-    self.assertEqual(
-      ('package/dir/foo',),
-      snapshot_paths
-    )
+    self.assertEqual(('package/dir/foo',), snapshot.files)
     self.assertEqual(target.sources_relative_to_target_base().files, ('foo',))
     self.assertEqual(target.sources_relative_to_buildroot(), ['package/dir/foo'])
diff --git a/tests/python/pants_test/task/test_simple_codegen_task.py b/tests/python/pants_test/task/test_simple_codegen_task.py
index 01d8143bd5a..3aecf4fd82f 100644
--- a/tests/python/pants_test/task/test_simple_codegen_task.py
+++ b/tests/python/pants_test/task/test_simple_codegen_task.py
@@ -14,6 +14,7 @@
 from pants.build_graph.target import Target
 from pants.task.simple_codegen_task import SimpleCodegenTask
 from pants.util.dirutil import safe_mkdtemp
+from pants.util.objects import datatype
 from pants_test.task_test_base import TaskTestBase, ensure_cached
 
 
@@ -113,6 +114,12 @@ def _copy_target_attributes(self):
     return ['copied']
 
 
+class DummyVersionedTarget(datatype(['target', 'results_dir'])):
+  @property
+  def current_results_dir(self):
+    return self.results_dir
+
+
 class SimpleCodegenTaskTest(TaskTestBase):
 
   @classmethod
@@ -212,12 +219,13 @@ def _do_test_duplication(self, targets, allow_dups, should_fail):
     def execute():
       for target in targets:
         target_workdir = target_workdirs[target]
+        vt = DummyVersionedTarget(target, target_workdir)
         task.execute_codegen(target, target_workdir)
-        sources = task._capture_sources(((target, target_workdir),))[0]
-        task._handle_duplicate_sources(target, target_workdir, sources)
+        sources = task._capture_sources((vt,))[0]
+        task._handle_duplicate_sources(vt, sources)
         # _handle_duplicate_sources may delete files from the filesystem, so we need to re-capture.
-        sources = task._capture_sources(((target, target_workdir),))[0]
-        syn_targets.append(task._inject_synthetic_target(target, target_workdir, sources))
+        sources = task._capture_sources((vt,))[0]
+        syn_targets.append(task._inject_synthetic_target(vt, sources))
 
     if should_fail:
       # If we're expected to fail, validate the resulting message.