# Process and execute instructions from the parse tree.

import abc
import ast
import glob
import json
import os
import os.path
import re
import shutil
import sys

import charliecloud as ch
import build_cache as bu
import filesystem as fs
import force
import image as im


## Globals ##

# ARG values that are set before FROM.
argfrom = {}

# Namespace from command line arguments. FIXME: be more tidy about this ...
cli = None

# --force injector object (initialized to something meaningful during FROM).
forcer = None

# Images that we are building. Each stage gets its own image. In this
# dictionary, an image appears exactly once or twice. All images appear with
# an int key counting stages up from zero. Images with a name (e.g., “FROM ...
# AS foo”) have a second string key of the name.
images = dict()

# Number of stages. This is obtained by counting FROM instructions in the
# parse tree, so we can use it for error checking.
image_ct = None

## Imports not in standard library ##

# See image.py for the messy import of this.
lark = im.lark

## Exceptions ##

class Instruction_Ignored(Exception): pass

## Main loop ##

class Environment:
   """The state we are in: environment variables, working directory, etc. Most
      of this is just passed through from the image metadata."""


# Class responsible for traversing the parse tree generated by lark. “Main_Loop”
# visits each node in the parse tree and calls its “__default__” method to
# figure out what to do with the node. This behavior is defined by the parent
# class, “lark.Visitor”, documented here:
# https://lark-parser.readthedocs.io/en/latest/visitors.html
class Main_Loop(lark.Visitor):

   __slots__ = ("instruction_total_ct",
                "miss_ct",    # number of misses during this stage
                "inst_prev")  # last instruction executed

   def __init__(self, *args, **kwargs):
      self.miss_ct = 0
      self.inst_prev = None
      self.instruction_total_ct = 0
      super().__init__(*args, **kwargs)

   # The tree parameter is the current node being visited and represents the
   # root node of the current subtree. The “tree.data” attribute holds the
   # name of the instruction such as “copy”. We append “_G” to form a class
   # name which implements the instruction’s grammar rules. When a
   # “lark.Visitor” visits a parse tree node, it calls the method which
   # matches the “tree.data” attribute. Since we don’t define these methods,
   # it falls back to “__default__” where we call on the corresponding “_G”
   # class to handle the instruction.
   def __default__(self, tree):
      class_ = tree.data.title() + "_G"
      if (class_ in globals()):
         inst = globals()[class_](tree)
         if (self.instruction_total_ct == 0):
            if (not (   isinstance(inst, Directive_G)
                     or isinstance(inst, From__G)
                     or isinstance(inst, Instruction_No_Image))):
               ch.FATAL("first instruction must be ARG or FROM")
         inst.init(self.inst_prev)
         # The three announce_maybe() calls are clunky but I couldn’t figure
         # out how to avoid the repeats.
         # Prepare instructions and execute if there’s a cache miss.
         try:
            self.miss_ct = inst.prepare(self.miss_ct)
            inst.announce_maybe()
         except Instruction_Ignored:
            inst.announce_maybe()
            return
         except ch.Fatal_Error:
            inst.announce_maybe()
            inst.prepare_rollback()
            raise
         if (inst.miss):
            if (self.miss_ct == 1):
               inst.checkout_for_build()
            try:
               inst.execute()
            except ch.Fatal_Error:
               inst.rollback()
               raise
            if (inst.image_i >= 0):
               inst.metadata_update()
            inst.commit()
         self.inst_prev = inst
         self.instruction_total_ct += 1

# Traverse the Lark parse tree and perform actions based on the instructions.
# Instruction nodes must be visited in order. The source code [1] shows that
# `visit_topdown()` uses `iter_subtrees_topdown()`, which guarantees that nodes are visited
# in the same order as how `pretty()` prints the tree [2].
# [1] https://github.com/lark-parser/lark/blob/445c8d4/lark/visitors.py#L211
# [2] https://lark-parser.readthedocs.io/en/latest/classes.html#lark.Tree.iter_subtrees_topdown\
def parse_tree_traverse(tree, image_ct_, cli_):
   global image_ct
   global cli
   image_ct = image_ct_
   cli = cli_

   ml = Main_Loop()
   # Use top-down traversal to visit instruction nodes in order.
   ml.visit_topdown(tree)
   if (ml.instruction_total_ct > 0):
      if (ml.miss_ct == 0):
         ml.inst_prev.checkout()
      ml.inst_prev.ready()

   # Check that all build arguments were consumed.
   if (len(cli.build_arg) != 0):
      ch.FATAL("--build-arg: not consumed: " + " ".join(cli.build_arg.keys()))

   # Print summary & we’re done.
   if (ml.instruction_total_ct == 0):
      ch.FATAL("no instructions found: %s" % cli.file)
   assert (ml.inst_prev.image_i + 1 == image_ct)  # should’ve errored already
   if ((cli.force != ch.Force_Mode.NONE) and ml.miss_ct != 0):
      ch.INFO("--force=%s: modified %d RUN instructions"
              % (cli.force.value, forcer.run_modified_ct))

   ch.INFO("grown in %d instructions: %s"
           % (ml.instruction_total_ct, ml.inst_prev.image))

   # Make sure entrypoint/cmd files are created
   ml.inst_prev.image.metadata_config_entrypoint_cmd()

   # FIXME: remove when we’re done encouraging people to use the build cache.
   if (isinstance(bu.cache, bu.Disabled_Cache)):
      ch.INFO("build slow? consider enabling the build cache",
              "https://hpc.github.io/charliecloud/command-usage.html#build-cache")

def unescape(sl):
   # FIXME: This is also ugly and should go in the grammar.
   #
   # The Dockerfile spec does not precisely define string escaping, but I’m
   # guessing it’s the Go rules. You will note that we are using Python rules.
   # This is wrong but close enough for now (see also gripe in previous
   # paragraph).
   if (    not sl.startswith('"')                          # no start quote
       and (not sl.endswith('"') or sl.endswith('\\"'))):  # no end quote
      sl = '"%s"' % sl
   assert (len(sl) >= 2 and sl[0] == '"' and sl[-1] == '"' and sl[-2:] != '\\"')
   return ast.literal_eval(sl)

## Supporting classes ##

class Instruction(abc.ABC):

   __slots__ = ("announced_p",
                "commit_files",  # modified files; default: anything
                "git_hash",      # Git commit where sid was found
                "image",
                "image_alias",
                "image_i",
                "lineno",
                "options",       # consumed
                "options_str",   # saved at instantiation
                "parent",
                "sid",
                "tree")

   def __init__(self, tree):
      """Note: When this is called, all we know about the instruction is
         what’s in the parse tree. In particular, you must not call
         ch.variables_sub() here."""
      self.announced_p = False
      self.commit_files = set()
      self.git_hash = bu.GIT_HASH_UNKNOWN
      self.lineno = tree.meta.line
      self.options = dict()
      # saving options with only 1 saved value
      for st in tree.children_("option"):
         k = st.terminal("OPTION_KEY")
         v = st.terminal("OPTION_VALUE")
         if (k in self.options):
            ch.FATAL("%3d %s: repeated option --%s"
                     % (self.lineno, self.str_name, k))
         self.options[k] = v

      # saving keypair options in a dictionary
      for st in tree.children_("option_keypair"):
         k = st.terminal("OPTION_KEY")
         s = st.terminal("OPTION_VAR")
         v = st.terminal("OPTION_VALUE")
         # assuming all key pair options allow multiple options
         self.options.setdefault(k, {}).update({s: v})

      ol = list()
      for (k, v) in self.options.items():
         if (isinstance(v, dict)):
            for (k2, v) in v.items():
               ol.append("--%s=%s=%s" % (k, k2, v))
         else:
            ol.append("--%s=%s" % (k, v))
      self.options_str = " ".join(ol)
      self.tree = tree
      # These are set in init().
      self.image = None
      self.parent = None
      self.image_alias = None
      self.image_i = None

   def __str__(self):
      options = self.options_str
      if (options != ""):
         options = " " + options
      return "%s%s %s" % (self.str_name, options, self.str_)

   @property
   def env_arg(self):
      if (self.image is None):
         assert False, "unimplemented"  # return dict()
      else:
         return self.image.metadata["arg"]

   @property
   def env_build(self):
      return { **self.env_arg, **self.env_env }

   @property
   def env_env(self):
      if (self.image is None):
         assert False, "unimplemented"  # return dict()
      else:
         return self.image.metadata["env"]

   @property
   def miss(self):
      """This is actually a three-valued property:

           1. True  => miss
           2. False => hit
           3. None  => unknown or n/a"""
      if (self.git_hash == bu.GIT_HASH_UNKNOWN):
         return None
      else:
         return (self.git_hash is None)

   @property
   def shell(self):
      if (self.image is None):
         assert False, "unimplemented"  # return ["/bin/false"]
      else:
         return self.image.metadata["shell"]

   @shell.setter
   def shell(self, x):
      self.image.metadata["shell"] = x

   @property
   def sid_input(self):
      return str(self).encode("UTF-8")

   @property
   def status_char(self):
      return bu.cache.status_char(self.miss)

   @property
   @abc.abstractmethod
   def str_(self):
      ...

   @property
   def str_name(self):
      return self.__class__.__name__.split("_")[0].upper()

   @property
   def workdir(self):
      return fs.Path(self.image.metadata["cwd"])

   @workdir.setter
   def workdir(self, x):
      self.image.metadata["cwd"] = str(x)

   def announce_maybe(self):
      "Announce myself if I haven’t already been announced."
      if (not self.announced_p):
         self_ = str(self)
         if (ch.user() == "qwofford" and sys.stderr.isatty()):
            self_ = re.sub(r"^RSYNC", "NSYNC", self_)
         ch.INFO("%3s%s %s" % (self.lineno, self.status_char, self_))
         self.announced_p = True

   def chdir(self, path):
      if (path.is_absolute()):
         self.workdir = path
      else:
         self.workdir //= path

   def checkout(self, base_image=None):
      bu.cache.checkout(self.image, self.git_hash, base_image)

   def checkout_for_build(self, base_image=None):
      self.parent.checkout(base_image)
      global forcer
      forcer = force.new(self.image.unpack_path, cli.force, cli.force_cmd)

   def commit(self):
      path = self.image.unpack_path
      self.git_hash = bu.cache.commit(path, self.sid, str(self),
                                      self.commit_files)

   def execute(self):
      """Do what the instruction says. At this point, the unpack directory is
         all ready to go. Thus, the method is cache-ignorant."""
      pass

   def init(self, parent):
      """Initialize attributes defining this instruction’s context, much of
         which is not available until the previous instruction is processed.
         After this is called, the instruction has a valid image and parent
         instruction, unless it’s the first instruction, in which case
         prepare() does the initialization."""
      # Separate from prepare() because subclasses shouldn’t need to override
      # it. If a subclass doesn’t like the result, it can just change things
      # in prepare().
      self.parent = parent
      if (self.parent is None):
         self.image_i = -1
      else:
         self.image = self.parent.image
         self.image_alias = self.parent.image_alias
         self.image_i = self.parent.image_i

   def metadata_update(self):
      self.image.metadata["history"].append(
         { "created": ch.now_utc_iso8601(),
           "created_by": "%s %s" % (self.str_name, self.str_)})
      self.image.metadata_save()

   def options_assert_empty(self):
      try:
         k = next(iter(self.options.keys()))
         ch.FATAL("%s: invalid option --%s" % (self.str_name, k))
      except StopIteration:
         pass

   def prepare(self, miss_ct):
      """Set up for execution; parent is the parent instruction and miss_ct is
         the number of misses in this stage so far. Returns the new number of
         misses; usually miss_ct if this instruction hit or miss_ct + 1 if it
         missed. Some instructions (e.g., FROM) resets the miss count.
         Announce self as soon as hit/miss status is known, hopefully before
         doing anything complicated or time-consuming.

         Typically, subclasses will set up enough state for self.sid_input to
         be valid, then call super().prepare().

         Gotchas:

           1. Announcing the instruction: Subclasses that are fast can let the
              caller announce. However, subclasses that consume non-trivial
              time in prepare() should call announce_maybe() as soon as they
              know hit/miss status.

           2. Errors: The caller catches Fatal_Error, announces, calls
              prepare_rollback(), and then re-raises. This to ensure the
              instruction is announced (see #1486) and any
              possibly-inconsistent state is fixed before existing.

           3. Modifying image metadata: Instructions like ARG, ENV, FROM,
              LABEL, SHELL, and WORKDIR must modify metadata here, not in
              execute(), so it’s available to later instructions even on
              cache hit."""
      self.sid = bu.cache.sid_from_parent(self.parent.sid, self.sid_input)
      self.git_hash = bu.cache.find_sid(self.sid, self.image.ref.for_path)
      return miss_ct + int(self.miss)

   def prepare_rollback(self):
      pass  # typically a no-op

   def ready(self):
      bu.cache.ready(self.image)

   def rollback(self):
      """Discard everything done by execute(), which may have completed
         partially, fully, or not at all."""
      bu.cache.rollback(self.image.unpack_path)

   def unsupported_forever_warn(self, msg):
      ch.WARNING("not supported, ignored: %s %s" % (self.str_name, msg))

   def unsupported_yet_fatal(self, msg, issue_no):
      ch.FATAL("not yet supported: issue #%d: %s %s"
               % (issue_no, self.str_name, msg))

   def unsupported_yet_warn(self, msg, issue_no):
      ch.WARNING("not yet supported, ignored: issue #%d: %s %s"
                 % (issue_no, self.str_name, msg))


class Copy(Instruction):

   # Superclass for instructions that do some flavor of file copying (ADD,
   # COPY, RSYNC).

   __slots__ = ("dst",          # string b/c trailing slash is significant
                "dst_raw",
                "from_",
                "src_metadata",
                "srcs",         # strings b/c trailing slashes are significant
                "srcs_base",
                "srcs_raw")

   @property
   def sid_input(self):
      return super().sid_input + self.src_metadata

   def expand_dest(self):
      """Set self.dst from self.dst_raw with environment variables expanded
         and image root prepended."""
      dst_raw = ch.variables_sub(self.dst_raw, self.env_build)
      if (len(dst_raw) < 1):
         ch.FATAL("destination is empty after expansion: %s" % self.dst_raw)
      base = self.image.unpack_path
      if (dst_raw[0] != "/"):
         base //= self.workdir
      self.dst = base // ch.variables_sub(self.dst_raw, self.env_build)

   def expand_sources(self):
      """Set self.srcs from self.srcs_raw with environment variables and globs
         expanded, absolute paths with appropriate base, and validate that
         they are within the sources base."""
      if (cli.context == "-" and self.from_ is None):
         ch.FATAL("no context because “-” given")
      if (len(self.srcs_raw) < 1):
         ch.FATAL("source or destination missing")
      self.srcs_base_set()
      self.srcs = list()
      for src in (ch.variables_sub(i, self.env_build) for i in self.srcs_raw):
         # glob can’t take Path
         matches = sorted(fs.Path(i)
                          for i in glob.glob("%s/%s" % (self.srcs_base, src)))
         if (len(matches) == 0):
            ch.FATAL("source not found: %s" % src)
         for m in matches:
            self.srcs.append(m)
            ch.VERBOSE("source: %s" % m)
            self.src_context_validate(m, src)

   def src_context_validate(self, m, src):
      """Validate source is within context directory. (We need the source as
         given later, so don’t canonicalize persistently.) There is no clear
         substitute for `commonpath()` in `pathlib`."""
      mc = m.resolve()
      if (not os.path.commonpath([mc, self.srcs_base])
                     .startswith(self.srcs_base)):
         ch.FATAL("can’t copy from outside context: %s" % src)

   def srcs_base_set(self):
      "Set self.srcs_base according to context and --from."
      if (self.from_ is None):
         self.srcs_base = cli.context
      else:
         if (self.from_ == self.image_i or self.from_ == self.image_alias):
            ch.FATAL("--from: stage %s is the current stage" % self.from_)
         if (not self.from_ in images):
            # FIXME: Would be nice to also report if a named stage is below.
            if (isinstance(self.from_, int) and self.from_ < image_ct):
               if (self.from_ < 0):
                  ch.FATAL("--from: invalid negative stage index %d"
                           % self.from_)
               else:
                  ch.FATAL("--from: stage %d does not exist yet"
                           % self.from_)
            else:
               ch.FATAL("--from: stage %s does not exist" % self.from_)
         self.srcs_base = images[self.from_].unpack_path
      self.srcs_base = os.path.realpath(self.srcs_base)
      ch.VERBOSE("context: %s" % self.srcs_base)


class Instruction_No_Image(Instruction):
   # This is a class for instructions that do not affect the image, i.e.,
   # no-op from the image’s perspective, but executed for their side effects,
   # e.g., changing some configuration. These instructions do not interact
   # with the build cache and can be executed when no image exists (i.e.,
   # before FROM).

   # FIXME: Only tested with instructions before the first FROM. I doubt it
   # works for instructions elsewhere.

   @property
   def miss(self):
      return True

   @property
   def status_char(self):
      return bu.cache.status_char(None)

   def checkout_for_build(self):
      pass

   def commit(self):
      pass

   def prepare(self, miss_ct):
      return miss_ct + int(self.miss)


class Instruction_Unsupported(Instruction):

   __slots__ = ()

   @property
   def miss(self):
      return None

   @property
   def str_(self):
      return "(unsupported)"


class Instruction_Supported_Never(Instruction_Unsupported):

   __slots__ = ()

   def prepare(self, *args):
      self.unsupported_forever_warn("instruction")
      raise Instruction_Ignored()


## Core classes ##

class Arg(Instruction):

   __slots__ = ("key",
                "value")

   def __init__(self, *args):
      super().__init__(*args)
      self.commit_files.add(fs.Path("ch/metadata.json"))
      self.key = self.tree.terminal("WORD", 0)
      if (self.key in cli.build_arg):
         self.value = cli.build_arg[self.key]
         del cli.build_arg[self.key]
      else:
         self.value = self.value_default()

   @property
   def sid_input(self):
      if (self.key in im.ARGS_MAGIC):
         return (self.str_name + self.key).encode("UTF-8")
      else:
         return super().sid_input

   @property
   def str_(self):
      s = "%s=" % self.key
      if (self.value is not None):
         s += "'%s'" % self.value
      if (self.key in im.ARGS_MAGIC):
         s += " [special]"
      return s

   def prepare(self, *args):
      if (self.value is not None):
         self.value = ch.variables_sub(self.value, self.env_build)
         self.env_arg[self.key] = self.value
      return super().prepare(*args)


class Arg_Bare_G(Arg):

   __slots__ = ()

   def value_default(self):
      return None


class Arg_Equals_G(Arg):

   __slots__ = ()

   def value_default(self):
      v = self.tree.terminal("WORD", 1)
      if (v is None):
         v = unescape(self.tree.terminal("STRING_QUOTED"))
      return v


class Arg_First(Instruction_No_Image):

   __slots__ = ("key",
                "value")

   def __init__(self, *args):
      super().__init__(*args)
      self.key = self.tree.terminal("WORD", 0)
      if (self.key in cli.build_arg):
         self.value = cli.build_arg[self.key]
         del cli.build_arg[self.key]
      else:
         self.value = self.value_default()

   @property
   def str_(self):
      s = "%s=" % self.key
      if (self.value is not None):
         s += "'%s'" % self.value
      if (self.key in im.ARGS_MAGIC):
         s += " [special]"
      return s

   def prepare(self, *args):
      if (self.value is not None):
         argfrom.update({self.key: self.value})
      return super().prepare(*args)


class Arg_First_Bare_G(Arg_First):

   __slots__ = ()

   def value_default(self):
      return None


class Arg_First_Equals_G(Arg_First):

   __slots__ = ()

   def value_default(self):
      v = self.tree.terminal("WORD", 1)
      if (v is None):
         v = unescape(self.tree.terminal("STRING_QUOTED"))
      return v

class CommandInstructions(Instruction):
   __slots__ = ("cmd",)

   @property
   def str_(self):
      return str(self.cmd)

   def execute(self):
      return super().execute()

   def prepare(self, miss_ct):
      if self.tree.child("string_list"):
         self.cmd = [ch.variables_sub(unescape(i), self.env_build)
                  for i in self.tree.child("string_list").terminals("STRING_QUOTED")]
      elif self.tree.child("line"):
         self.cmd = ch.variables_sub(self.tree.child("line").terminals_cat("LINE_CHUNK"),
                                 self.env_build)
      ch.VERBOSE("CommandInstructions: %s" % self.cmd)
      self.image.metadata[self.instruction] = self.cmd
      return super().prepare(miss_ct)

class Cmd_G(CommandInstructions):
   __slots__ = ()
   instruction = "cmd"

class Copy_G(Copy):

   # ABANDON ALL HOPE YE WHO ENTER HERE
   #
   # Note: The Dockerfile specification for COPY is complex, messy,
   # inexplicably different from cp(1), and incomplete. We try to be
   # bug-compatible with Docker (legacy builder, not BuildKit -- yes, they are
   # different) but probably are not 100%. See the FAQ.
   #
   # Because of these weird semantics, none of this abstracted into a general
   # copy function. I don’t want people calling it except from here.

   __slots__ = ()

   def __init__(self, *args):
      super().__init__(*args)
      self.from_ = self.options.pop("from", None)
      if (self.from_ is not None):
         try:
            self.from_ = int(self.from_)
         except ValueError:
            pass
      # No subclasses, so check what parse tree matched.
      if (self.tree.child("copy_shell") is not None):
         args = list(self.tree.child_terminals("copy_shell", "WORD"))
      elif (self.tree.child("copy_list") is not None):
         args = list(self.tree.child_terminals("copy_list", "STRING_QUOTED"))
         for i in range(len(args)):
            args[i] = args[i][1:-1]  # strip quotes
      else:
         assert False, "unreachable code reached"
      self.srcs_raw = args[:-1]
      self.dst_raw = args[-1]

   @property
   def str_(self):
      dst = repr(self.dst) if hasattr(self, "dst") else self.dst_raw
      return "%s -> %s" % (self.srcs_raw, dst)

   def copy_src_dir(self, src, dst):
      """Copy the contents of directory src, named by COPY, either explicitly
         or with wildcards, to dst. src might be a symlink, but dst is a
         canonical path. Both must be at the top level of the COPY
         instruction; i.e., this function must not be called recursively. dst
         must exist already and be a directory. Unlike subdirectories, the
         metadata of dst will not be altered to match src."""
      def onerror(x):
         ch.FATAL("can’t scan directory: %s: %s" % (x.filename, x.strerror))
      # Use Path objects in this method because the path arithmetic was
      # getting too hard with strings.
      src = src.resolve()  # alternative to os.path.realpath()
      dst = fs.Path(dst)
      assert (src.is_dir() and not src.is_symlink())
      assert (dst.is_dir() and not dst.is_symlink())
      ch.DEBUG("copying named directory: %s -> %s" % (src, dst))
      for (dirpath, dirnames, filenames) in ch.walk(src, onerror=onerror):
         subdir = dirpath.relative_to(src)
         dst_dir = dst // subdir
         # dirnames can contain symlinks, which we handle as files, so we’ll
         # rebuild it; the walk will not descend into those “directories”.
         dirnames2 = dirnames.copy()  # shallow copy
         dirnames[:] = list()         # clear in place
         for d in dirnames2:
            src_path = dirpath // d
            dst_path = dst_dir // d
            ch.TRACE("dir: %s -> %s" % (src_path, dst_path))
            if (os.path.islink(src_path)):
               filenames.append(d)  # symlink, handle as file
               ch.TRACE("symlink to dir, will handle as file")
               continue
            else:
               dirnames.append(d)   # directory, descend into later
            # If destination exists, but isn’t a directory, remove it.
            if (os.path.exists(dst_path)):
               if (os.path.isdir(dst_path) and not os.path.islink(dst_path)):
                  ch.TRACE("dst_path exists and is a directory")
               else:
                  ch.TRACE("dst_path exists, not a directory, removing")
                  dst_path.unlink()
            # If destination directory doesn’t exist, create it.
            if (not os.path.exists(dst_path)):
               ch.TRACE("mkdir dst_path")
               ch.ossafe("can’t mkdir: %s" % dst_path, os.mkdir, dst_path)
            # Copy metadata, now that we know the destination exists and is a
            # directory.
            ch.ossafe("can’t copy metadata: %s -> %s" % (src_path, dst_path),
                      shutil.copystat, src_path, dst_path, follow_symlinks=False)
         for f in filenames:
            src_path = dirpath // f
            dst_path = dst_dir // f
            ch.TRACE("file or symlink via copy2: %s -> %s"
                      % (src_path, dst_path))
            if (not (os.path.isfile(src_path) or os.path.islink(src_path))):
               ch.FATAL("can’t COPY: unknown file type: %s" % src_path)
            if (os.path.exists(dst_path)):
               ch.TRACE("destination exists, removing")
               if (os.path.isdir(dst_path) and not os.path.islink(dst_path)):
                  dst_path.rmtree()
               else:
                  dst_path.unlink()
            src_path.copy(dst_path)

   def copy_src_file(self, src, dst):
      """Copy file src to dst. src might be a symlink, but dst is a canonical
         path. Both must be at the top level of the COPY instruction; i.e.,
         this function must not be called recursively. dst has additional
         constraints:

           1. If dst is a directory that exists, src will be copied into that
              directory like cp(1); e.g. “COPY file_ /dir_” will produce a
              file in the imaged called. “/dir_/file_”.

           2. If dst is a regular file that exists, src will overwrite it.

           3. If dst is another type of file that exists, that’s an error.

           4. If dst does not exist, the parent of dst must be a directory
              that exists."""
      assert (src.is_file())
      assert (not dst.is_symlink())
      assert (   (dst.exists() and (dst.is_dir() or dst.is_file()))
              or (not dst.exists() and dst.parent.is_dir()))
      if (dst.is_dir()):
         dst //= src.name
      src = src.resolve()
      ch.DEBUG("copying named file: %s -> %s" % (src, dst))
      src.copy(dst)

   def dest_realpath(self, unpack_path, dst):
      """Return the canonicalized version of path dst within (canonical) image
         path unpack_path. We can’t use os.path.realpath() because if dst is
         an absolute symlink, we need to use the *image’s* root directory, not
         the host. Thus, we have to resolve symlinks manually."""
      dst_canon = unpack_path
      dst_parts = list(reversed(dst.parts))  # easier to operate on end of list
      iter_ct = 0
      while (len(dst_parts) > 0):
         iter_ct += 1
         if (iter_ct > 100):  # arbitrary
            ch.FATAL("can’t COPY: too many path components")
         ch.TRACE("current destination: %d %s" % (iter_ct, dst_canon))
         #ch.TRACE("parts remaining: %s" % dst_parts)
         part = dst_parts.pop()
         if (part == "/" or part == "//"):  # 3 or more slashes yields "/"
            ch.TRACE("skipping root")
            continue
         cand = dst_canon // part
         ch.TRACE("checking: %s" % cand)
         if (not cand.is_symlink()):
            ch.TRACE("not symlink")
            dst_canon = cand
         else:
            target = fs.Path(os.readlink(cand))
            ch.TRACE("symlink to: %s" % target)
            assert (len(target.parts) > 0)  # POSIX says no empty symlinks
            if (target.is_absolute()):
               ch.TRACE("absolute")
               dst_canon = fs.Path(unpack_path)
            else:
               ch.TRACE("relative")
            dst_parts.extend(reversed(target.parts))
      return dst_canon

   def execute(self):
      # Locate the destination.
      unpack_canon = fs.Path(self.image.unpack_path).resolve()
      if (self.dst.startswith("/")):
         dst = fs.Path(self.dst)
      else:
         dst = self.workdir // self.dst
      ch.VERBOSE("destination, as given: %s" % dst)
      dst_canon = self.dest_realpath(unpack_canon, dst) # strips trailing slash
      ch.VERBOSE("destination, canonical: %s" % dst_canon)
      if (not os.path.commonpath([dst_canon, unpack_canon])
              .startswith(str(unpack_canon))):
         ch.FATAL("can’t COPY: destination not in image: %s" % dst_canon)
      # Create the destination directory if needed.
      if (   self.dst.endswith("/")
          or len(self.srcs) > 1
          or self.srcs[0].is_dir()):
         if (not dst_canon.exists()):
            dst_canon.mkdirs()
         elif (not dst_canon.is_dir()):  # not symlink b/c realpath()
            ch.FATAL("can’t COPY: not a directory: %s" % dst_canon)
      if (dst_canon.parent.exists()):
         if (not dst_canon.parent.is_dir()):
            ch.FATAL("can’t COPY: not a directory: %s" % dst_canon.parent)
      else:
         dst_canon.parent.mkdirs()
      # Copy each source.
      for src in self.srcs:
         if (src.is_file()):
            self.copy_src_file(src, dst_canon)
         elif (src.is_dir()):
            self.copy_src_dir(src, dst_canon)
         else:
            ch.FATAL("can’t COPY: unknown file type: %s" % src)

   def prepare(self, miss_ct):
      # Complain about unsupported stuff.
      if (self.options.pop("chown", False)):
         self.unsupported_forever_warn("--chown")
      # Any remaining options are invalid.
      self.options_assert_empty()
      # Expand operands.
      self.expand_sources()
      self.dst = ch.variables_sub(self.dst_raw, self.env_build)
      # Gather metadata for hashing.
      self.src_metadata = fs.Path.stat_bytes_all(self.srcs)
      # Pass on to superclass.
      return super().prepare(miss_ct)

class Directive_G(Instruction_Supported_Never):

   __slots__ = ()

   @property
   def str_name(self):
      return "#%s" % self.tree.terminal("DIRECTIVE_NAME")

   def prepare(self, *args):
      ch.WARNING("not supported, ignored: parser directives")
      raise Instruction_Ignored()

class Entrypoint_G(CommandInstructions):
   __slots__ = ()
   instruction = "entrypoint"

class Env(Instruction):

   __slots__ = ("key",
                "value")

   def __init__(self, *args):
      super().__init__(*args)
      self.commit_files |= {fs.Path("ch/environment"),
                            fs.Path("ch/metadata.json")}

   @property
   def str_(self):
      return "%s='%s'" % (self.key, self.value)

   def execute(self):
      with (self.image.unpack_path // "/ch/environment").open("wt") as fp:
         for (k, v) in self.env_env.items():
            print("%s=%s" % (k, v), file=fp)

   def prepare(self, *args):
      self.value = ch.variables_sub(unescape(self.value), self.env_build)
      self.env_env[self.key] = self.value
      return super().prepare(*args)


class Env_Equals_G(Env):

   __slots__ = ()

   def __init__(self, *args):
      super().__init__(*args)
      self.key = self.tree.terminal("WORD", 0)
      self.value = self.tree.terminal("WORD", 1)
      if (self.value is None):
         self.value = self.tree.terminal("STRING_QUOTED")


class Env_Space_G(Env):

   __slots__ = ()

   def __init__(self, *args):
      super().__init__(*args)
      self.key = self.tree.terminal("WORD")
      self.value = self.tree.terminals_cat("LINE_CHUNK")


class From__G(Instruction):

   __slots__ = ("alias",
                "base_alias",
                "base_image",
                "base_text")

   # Not meaningful for FROM.
   sid_input = None

   def __init__(self, *args):
      super().__init__(*args)
      argfrom.update(self.options.pop("arg", {}))

   @property
   def str_(self):
      if (hasattr(self, "base_alias")):
         base_text = str(self.base_alias)
      elif (hasattr(self, "base_image")):
         base_text = str(self.base_image.ref)
      else:
         # Initialization failed, but we want to print *something*.
         base_text = self.base_text
      return base_text + ((" AS " + self.alias) if self.alias else "")

   def checkout_for_build(self):
      assert (isinstance(bu.cache, bu.Disabled_Cache))
      super().checkout_for_build(self.base_image)

   def execute(self):
      # Everything happens in prepare().
      pass

   def metadata_update(self, *args):
      # FROM doesn’t update metadata because it never misses when the cache is
      # enabled, so this would never be called, and we want disabled results
      # to be the same. In particular, FROM does not generate history entries.
      pass

   def prepare(self, miss_ct):
      # FROM is special because its preparation involves opening a new stage
      # and closing the previous if there was one. Because of this, the actual
      # parent is the last instruction of the base image.
      self.base_text = self.tree.child_terminals_cat("image_ref", "IMAGE_REF")
      self.alias = self.tree.child_terminal("from_alias", "IR_PATH_COMPONENT")
      # Validate instruction.
      if (self.options.pop("platform", False)):
         self.unsupported_yet_fatal("--platform", 778)
      self.options_assert_empty()
      # Update context.
      self.image_i += 1
      self.image_alias = self.alias
      if (self.image_i == image_ct - 1):
         # Last image; use tag unchanged.
         tag = cli.tag
      elif (self.image_i > image_ct - 1):
         # Too many images!
         ch.FATAL("expected %d stages but found at least %d"
                  % (image_ct, self.image_i + 1))
      else:
         # Not last image; append stage index to tag.
         tag = "%s_stage%d" % (cli.tag, self.image_i)
      if self.base_text in images:
         # Is alias; store base_text as the “alias used” to target a previous
         # stage as the base.
         self.base_alias = self.base_text
         self.base_text = str(images[self.base_text].ref)
      self.base_image = im.Image(im.Reference(self.base_text, argfrom))
      self.image = im.Image(im.Reference(tag))
      images[self.image_i] = self.image
      if (self.image_alias is not None):
         images[self.image_alias] = self.image
      ch.VERBOSE("image path: %s" % self.image.unpack_path)
      # More error checking.
      if (str(self.image.ref) == str(self.base_image.ref)):
         ch.FATAL("output image ref same as FROM: %s" % self.base_image.ref)
      # Close previous stage if needed. In particular, we need the previous
      # stage’s image directory to exist because (a) we need to read its
      # metadata and (b) in case there’s a COPY later. Cache disabled will
      # already have the image directory and there is no notion of branch
      # “ready”, so do nothing in that case.
      if (self.image_i > 0 and not isinstance(bu.cache, bu.Disabled_Cache)):
         if (miss_ct == 0):
            # No previous miss already checked out the image. This will still
            # be fast most of the time since the correct branch is likely
            # checked out already.
            self.parent.checkout()
         self.parent.ready()
      # At this point any meaningful parent of FROM, e.g., previous stage, has
      # been closed; thus, act as own parent.
      self.parent = self
      # Pull base image if needed. This tells us hit/miss.
      (self.sid, self.git_hash) = bu.cache.find_image(self.base_image)
      unpack_no_git = (    self.base_image.unpack_exist_p
                       and not self.base_image.unpack_cache_linked)
      # Announce (before we start pulling).
      self.announce_maybe()
      # FIXME: shouldn’t know or care whether build cache is enabled here.
      if (self.miss):
         if (unpack_no_git):
            # Use case is mostly images built by old ch-image still in storage.
            if (not isinstance(bu.cache, bu.Disabled_Cache)):
               ch.WARNING("base image only exists non-cached; adding to cache")
            (self.sid, self.git_hash) = bu.cache.adopt(self.base_image)
         else:
            (self.sid, self.git_hash) = bu.cache.pull_lazy(self.base_image,
                                                           self.base_image.ref)
      elif (unpack_no_git):
         ch.WARNING("base image also exists non-cached; using cache")
      # Load metadata
      self.image.metadata_load(self.base_image)
      self.env_arg.update(argfrom)  # from pre-FROM ARG

      # Done.
      return int(self.miss)  # will still miss in disabled mode

   def prepare_rollback(self):
      # AFAICT the only thing that might be busted is the unpack directories
      # for either the base image or the image. We could probably be smarter
      # about this, but for now just delete them.
      try:
         base_image = self.base_image
      except AttributeError:
         base_image = None
      try:
         image = self.image
      except AttributeError:
         image = None
      if (base_image is not None or image is not None):
         ch.INFO("something went wrong, rolling back ...")
         if (base_image is not None):
            bu.cache.unpack_delete(base_image, missing_ok=True)
         if (image is not None):
            bu.cache.unpack_delete(image, missing_ok=True)


class Label(Instruction):

   __slots__ = ("key",
                "value")

   def __init__(self, *args):
      super().__init__(*args)
      self.commit_files |= {ch.Path("ch/metadata.json")}

   @property
   def str_(self):
      return "%s='%s'" % (self.key, self.value)

   def prepare(self, *args):
      self.value = ch.variables_sub(unescape(self.value), self.env_build)
      self.image.metadata["labels"][self.key] = self.value
      return super().prepare(*args)


class Label_Equals_G(Label):

   __slots__ = ()

   def __init__(self, *args):
      super().__init__(*args)
      self.key = self.tree.terminal("WORD", 0)
      self.value = self.tree.terminal("WORD", 1)
      if (self.value is None):
         self.value = self.tree.terminal("STRING_QUOTED")


class Label_Space_G(Label):

   __slots__ = ()

   def __init__(self, *args):
      super().__init__(*args)
      self.key = self.tree.terminal("WORD")
      self.value = self.tree.terminals_cat("LINE_CHUNK")


class Rsync_G(Copy):

   __slots__ = ("plus_option",
                "rsync_options")

   def __init__(self, *args):
      super().__init__(*args)
      self.from_ = None  # not supported yet
      line_no = self.tree.meta.line
      st = self.tree.child("option_plus")
      self.plus_option = "l" if st is None else st.terminal("OPTION_LETTER")
      options_done = False
      self.rsync_options = list()
      self.srcs_raw = list()
      for word in self.tree.terminals("WORDE"):
         if (not options_done and word.startswith("-")):
            # Option. See assumption in docs that makes parsing a lot easier.
            if (word == "--"):             # end of options
               options_done = True
            elif (word.startswith("--")):  # long option
               self.rsync_options.append(word)
            else:                          # short option(s)
               if (len(word) == 1):
                  ch.FATAL("RSYNC: %d: invalid argument: %s" % (line_no, word))
               # Append options individually so we can process them more later.
               for m in re.finditer(r"[^=]=.*$|[^=]", word[1:]):
                  self.rsync_options.append("-" + m[0])
            continue
         # Not an option, so it must be a source or destination path.
         self.srcs_raw.append(word)
      if (len(self.srcs_raw) == 0):
         ch.FATAL("RSYNC: %d: source and destination missing" % line_no)
      self.dst_raw = self.srcs_raw.pop()

   @property
   def rsync_options_concise(self):
      "Return self.rsync_options with short options coalesced."
      # We don’t group short options with an argument even though we could
      # because it seems confusing, e.g. “-ab=c” vs. “-a -b=c”.
      def ship_out():
         nonlocal group
         if (group != ""):
            ret.append(group)
            group = ""
      ret = list()
      group = ""
      for o in self.rsync_options:
         if (o.startswith("--")):  # long option, not grouped
            ship_out()
            ret.append(o)
         elif (len(o) > 2):        # short option with argument, not grouped
            ship_out()
            ret.append(o)
         else:                     # short option without argument, grouped
            if (group == ""):
               group = "-"
            group += o[1:]         # add to group
      ship_out()
      return ret

   @property
   def str_(self):
      ret = list()
      if (self.plus_option is not None):
         ret.append("+" + self.plus_option)
      if (len(self.rsync_options_concise) > 0):
         ret += self.rsync_options_concise
      ret += self.srcs_raw
      ret.append(self.dst_raw)
      return " ".join(ret)

   def execute(self):
      plus_options = list()
      if (self.plus_option in "lmu"):  # no action needed for +z
         # see man page for explanations
         plus_options = ["-@=-1", "-AHSXpr"]
         if (sys.stderr.isatty()):
            plus_options += ["--info=progress2"]
         if (self.plus_option == "l"):
            plus_options += ["-l", "--safe-links"]
         elif (self.plus_option == "u"):
            plus_options += ["-l", "--copy-unsafe-links"]
      ch.cmd(["rsync"] + plus_options + self.rsync_options_concise
                       + self.srcs + [self.dst])

   def expand_rsync_froms(self):
      for i in range(len(self.rsync_options)):
         o = self.rsync_options[i]
         m = re.search("^--([a-z]+)-from=(.+)$", o)
         if (m is not None):
            key = m[1]
            if (m[2] == "-"):
               ch.FATAL("--*-from: can’t use standard input")
            elif (":" in m[2]):
               ch.FATAL("--*-from: can’t use remote hosts (colon in path)")
            path = ch.Path(m[2])
            if (path.is_absolute()):
               path = self.image.unpack_path // path
            else:
               path = self.srcs_base // path
            self.rsync_options[i] = "--%s-from=%s" % (key, path)

   def prepare(self, miss_ct):
      self.rsync_validate()
      # Expand operands.
      self.expand_sources()
      self.expand_dest()
      self.expand_rsync_froms()
      # Gather metadata for hashing.
      self.src_metadata = fs.Path.stat_bytes_all(self.srcs)
      # Pass on to superclass.
      return super().prepare(miss_ct)

   def rsync_validate(self):
      # Reject bad + options.
      if (self.plus_option not in ("mluz")):
         ch.FATAL("invalid plus option: %s" % self.plus_option)
      # Reject SSH and rsync transports. I *believe* simply the presence of
      # “:” (colon) in the filename triggers this behavior.
      for src in self.srcs_raw:
         if (":" in src):
            ch.FATAL("SSH and rsync transports not supported: %s" % src)
      # Reject bad flags.
      bad = { "--daemon",
              "-n", "--dry-run",
              "--remove-source-files" }
      for o in self.rsync_options:
         if (o in bad):
            ch.FATAL("disallowed option: %s" % o)

   def src_context_validate(self, *args):
      """We let rsync(1) enforce in-contextness because in some cases it
         actually can reach outside the context directory."""
      pass


class Run(Instruction):

   __slots__ = ("cmd")

   @property
   def str_name(self):
      # Can’t get this from the forcer object because it might not have been
      # initialized yet.
      if (cli.force == ch.Force_Mode.NONE):
         tag = ".N"
      elif (cli.force == ch.Force_Mode.FAKEROOT):
         # FIXME: This causes spurious misses because it adds the force tag to
         # *all* RUN instructions, not just those that actually were modified
         # (i.e, any RUN instruction will miss the equivalent RUN without
         # --force=fakeroot). But we don’t know know if an instruction needs
         # modifications until the result is checked out, which happens after
         # we check the cache. See issue #1339.
         tag = ".F"
      elif (cli.force == ch.Force_Mode.SECCOMP):
         tag = ".S"
      else:
         assert False, "unreachable code reached (force mode = %s)" % cli.force
      return super().str_name + tag

   def execute(self):
      rootfs = self.image.unpack_path
      cmd = forcer.run_modified(self.cmd, self.env_build)
      exit_code = ch.ch_run_modify(rootfs, cmd, self.env_build, self.workdir,
                                   cli.bind, forcer.ch_run_args, fail_ok=True)
      if (exit_code != 0):
         ch.FATAL("build failed: RUN command exited with %d" % exit_code)


class Run_Exec_G(Run):

   __slots__ = ()

   @property
   def str_(self):
      return json.dumps(self.cmd)  # double quotes, shlex.quote is less verbose

   def prepare(self, *args):
      self.cmd = [    ch.variables_sub(unescape(i), self.env_build)
                  for i in self.tree.terminals("STRING_QUOTED")]
      return super().prepare(*args)


class Run_Shell_G(Run):

   # Note re. line continuations and whitespace: Whitespace before the
   # backslash is passed verbatim to the shell, while the newline and any
   # whitespace between the newline and baskslash are deleted.

   __slots__ = ("_str_")

   @property
   def str_(self):
      return self._str_  # can’t replace abstract property with attribute

   def prepare(self, *args):
      cmd = self.tree.terminals_cat("LINE_CHUNK")
      self.cmd = self.shell + [cmd]
      self._str_ = cmd
      return super().prepare(*args)


class Shell_G(Instruction):

   def __init__(self, *args):
      super().__init__(*args)
      self.commit_files.add(fs.Path("ch/metadata.json"))

   @property
   def str_(self):
      return str(self.shell)

   def prepare(self, *args):
      self.shell = [    ch.variables_sub(unescape(i), self.env_build)
                    for i in self.tree.terminals("STRING_QUOTED")]
      return super().prepare(*args)


class Uns_Forever_G(Instruction_Supported_Never):

   __slots__ = ("name")

   def __init__(self, *args):
      super().__init__(*args)
      self.name = self.tree.terminal("UNS_FOREVER")

   @property
   def str_name(self):
      return self.name


class Uns_Yet_G(Instruction_Unsupported):

   __slots__ = ("issue_no",
                "name")

   def __init__(self, *args):
      super().__init__(*args)
      self.name = self.tree.terminal("UNS_YET")
      self.issue_no = { "ADD":         782,
                        "CMD":         780,
                        "ENTRYPOINT":  780,
                        "ONBUILD":     788 }[self.name]

   @property
   def str_name(self):
      return self.name

   def prepare(self, *args):
      self.unsupported_yet_warn("instruction", self.issue_no)
      raise Instruction_Ignored()


class Workdir_G(Instruction):

   __slots__ = ("path")

   @property
   def str_(self):
      return str(self.path)

   def execute(self):
      (self.image.unpack_path // self.workdir).mkdirs()

   def prepare(self, *args):
      self.path = fs.Path(ch.variables_sub(
         self.tree.terminals_cat("LINE_CHUNK"), self.env_build))
      self.chdir(self.path)
      return super().prepare(*args)