diff --git a/CHANGES.md b/CHANGES.md
index fd358937f64..7f1633d344e 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -20,10 +20,16 @@ ones in. -->
 
 ### Enhancements
 
+[#5229](https://github.com/cylc/cylc-flow/pull/5229) -
+- Added a single command to validate a previously run workflow against changes
+  to its source and reinstall a workflow.
+- Allows Cylc commands (including validate, list, view, config, and graph) to load template variables
+  configured by `cylc install` and `cylc play`.
+
 [#5184](https://github.com/cylc/cylc-flow/pull/5184) - scan for active
 runs of the same workflow at install time.
 
-[#5094](https://github.com/cylc/cylc-flow/pull/5094) - Added a single
+[#5121](https://github.com/cylc/cylc-flow/pull/5121) - Added a single
 command to validate, install and play a workflow.
 
 [#5032](https://github.com/cylc/cylc-flow/pull/5032) - set a default limit of
diff --git a/cylc/flow/option_parsers.py b/cylc/flow/option_parsers.py
index d64365cf20f..e9145010b75 100644
--- a/cylc/flow/option_parsers.py
+++ b/cylc/flow/option_parsers.py
@@ -80,9 +80,16 @@ def __init__(self, argslist, sources=None, useif=None, **kwargs):
             self.kwargs.update({kwarg: value})
 
     def __eq__(self, other):
-        """Args and Kwargs, but not other props equal."""
+        """Args and Kwargs, but not other props equal.
+
+        (Also make an exception for kwargs['help'] to allow lists of sources
+        prepended to 'help' to be passed through.)
+        """
         return (
-            self.kwargs == other.kwargs
+            (
+                {k: v for k, v in self.kwargs.items() if k != 'help'}
+                == {k: v for k, v in other.kwargs.items() if k != 'help'}
+            )
             and self.args == other.args
         )
 
@@ -117,6 +124,22 @@ def _update_sources(self, other):
     dest="icp"
 )
 
+AGAINST_SOURCE_OPTION = OptionSettings(
+    ['--against-source'],
+    help=(
+        "Load the workflow configuration from the source directory it was"
+        " installed from using any options (e.g. template variables) which"
+        " have been set in the installation."
+        " This is useful if you want to see how changes made to the workflow"
+        " source would affect the installation if reinstalled."
+        " Note if this option is used the provided workflow must have been"
+        " installed by `cylc install`."
+    ),
+    dest='against_source',
+    action='store_true',
+    default=False
+)
+
 
 icp_option = Option(
     *ICP_OPTION.args, **ICP_OPTION.kwargs)  # type: ignore[arg-type]
@@ -713,9 +736,10 @@ def combine_options_pair(first_list, second_list):
                 and first & second
             ):
                 # if any of the args are different:
+
                 if first.args == second.args:
-                    # if none of the arg names are different.
-                    raise Exception(f'Clashing Options \n{first}\n{second}')
+                    raise Exception(
+                        f'Clashing Options \n{first.args}\n{second.args}')
                 else:
                     first_args = first - second
                     second.args = second - first
diff --git a/cylc/flow/parsec/fileparse.py b/cylc/flow/parsec/fileparse.py
index 510a46d6322..1c1d34738ba 100644
--- a/cylc/flow/parsec/fileparse.py
+++ b/cylc/flow/parsec/fileparse.py
@@ -31,6 +31,7 @@
 """
 
 import os
+from optparse import Values
 from pathlib import Path
 import re
 import sys
@@ -45,6 +46,9 @@
 from cylc.flow.parsec.OrderedDict import OrderedDictWithDefaults
 from cylc.flow.parsec.include import inline
 from cylc.flow.parsec.util import itemstr
+from cylc.flow.templatevars import get_template_vars_from_db
+from cylc.flow.workflow_files import (
+    get_workflow_source_dir, check_flow_file)
 
 
 # heading/sections can contain commas (namespace name lists) and any
@@ -343,6 +347,43 @@ def merge_template_vars(
         return native_tvars
 
 
+def _prepend_old_templatevars(fpath: str, template_vars: t.Dict) -> t.Dict:
+    """If the fpath is in a rundir, extract template variables from database.
+
+    Args:
+        fpath: filepath of workflow config file.
+        template_vars: Template vars to prepend old tvars to.
+    """
+    rundir = Path(fpath).parent
+    old_tvars = get_template_vars_from_db(rundir)
+    old_tvars.update(template_vars)
+    template_vars = old_tvars
+    return template_vars
+
+
+def _get_fpath_for_source(fpath: str, opts: "Values") -> str:
+    """If `--against-source` is set and a sourcedir can be found, return
+    sourcedir.
+    Else return fpath unchanged.
+
+    Raises:
+        ParsecError: If validate against source is set and this is not
+        an installed workflow then we don't want to continue.
+    """
+    if getattr(opts, 'against_source', False):
+        thispath = Path(fpath)
+        source_dir = get_workflow_source_dir(thispath.parent)[0]
+        if source_dir:
+            retpath = check_flow_file(source_dir)
+            return str(retpath)
+        else:
+            raise ParsecError(
+                f'Cannot validate {thispath.parent} against source: '
+                'it is not an installed workflow.')
+    else:
+        return fpath
+
+
 def read_and_proc(
     fpath: str,
     template_vars: t.Optional[t.Dict[str, t.Any]] = None,
@@ -355,7 +396,9 @@ def read_and_proc(
     Jinja2 processing must be done before concatenation - it could be
     used to generate continuation lines.
     """
-
+    template_vars = template_vars if template_vars is not None else {}
+    template_vars = _prepend_old_templatevars(fpath, template_vars)
+    fpath = _get_fpath_for_source(fpath, opts)
     fdir = os.path.dirname(fpath)
 
     # Allow Python modules in lib/python/ (e.g. for use by Jinja2 filters).
diff --git a/cylc/flow/scripts/config.py b/cylc/flow/scripts/config.py
index 4907ae77176..aa7144a934d 100755
--- a/cylc/flow/scripts/config.py
+++ b/cylc/flow/scripts/config.py
@@ -49,14 +49,17 @@
   $ cylc config --initial-cycle-point=now myflow
 """
 
+import asyncio
+
 import os.path
 from typing import List, Optional, TYPE_CHECKING
 
 from cylc.flow.cfgspec.glbl_cfg import glbl_cfg
 from cylc.flow.config import WorkflowConfig
-from cylc.flow.id_cli import parse_id
+from cylc.flow.id_cli import parse_id_async
 from cylc.flow.exceptions import InputError
 from cylc.flow.option_parsers import (
+    AGAINST_SOURCE_OPTION,
     WORKFLOW_ID_OR_PATH_ARG_DOC,
     CylcOptionParser as COP,
     icp_option,
@@ -128,6 +131,9 @@ def get_option_parser() -> COP:
         action='store_true', default=False, dest='print_platforms'
     )
 
+    parser.add_option(
+        *AGAINST_SOURCE_OPTION.args, **AGAINST_SOURCE_OPTION.kwargs)
+
     parser.add_cylc_rose_options()
 
     return parser
@@ -149,6 +155,14 @@ def main(
     options: 'Values',
     *ids,
 ) -> None:
+    asyncio.run(_main(parser, options, *ids))
+
+
+async def _main(
+    parser: COP,
+    options: 'Values',
+    *ids,
+) -> None:
 
     if options.print_platform_names and options.print_platforms:
         options.print_platform_names = False
@@ -178,7 +192,7 @@ def main(
         )
         return
 
-    workflow_id, _, flow_file = parse_id(
+    workflow_id, _, flow_file = await parse_id_async(
         *ids,
         src=True,
         constraint='workflows',
diff --git a/cylc/flow/scripts/cylc.py b/cylc/flow/scripts/cylc.py
index 74a37119043..bd85cfa8e89 100644
--- a/cylc/flow/scripts/cylc.py
+++ b/cylc/flow/scripts/cylc.py
@@ -202,7 +202,8 @@ def get_version(long=False):
     'shutdown': 'stop',
     'task-message': 'message',
     'unhold': 'release',
-    'validate-install-play': 'vip'
+    'validate-install-play': 'vip',
+    'validate-reinstall': 'vr',
 }
 
 
diff --git a/cylc/flow/scripts/graph.py b/cylc/flow/scripts/graph.py
index 3488177ac17..ce22cfdc2ed 100644
--- a/cylc/flow/scripts/graph.py
+++ b/cylc/flow/scripts/graph.py
@@ -35,6 +35,7 @@
     $ cylc graph one -o 'one.svg'
 """
 
+import asyncio
 from difflib import unified_diff
 from shutil import which
 from subprocess import Popen, PIPE
@@ -45,8 +46,9 @@
 from cylc.flow.config import WorkflowConfig
 from cylc.flow.exceptions import InputError, CylcError
 from cylc.flow.id import Tokens
-from cylc.flow.id_cli import parse_id
+from cylc.flow.id_cli import parse_id_async
 from cylc.flow.option_parsers import (
+    AGAINST_SOURCE_OPTION,
     WORKFLOW_ID_OR_PATH_ARG_DOC,
     CylcOptionParser as COP,
     icp_option,
@@ -108,9 +110,10 @@ def get_nodes_and_edges(
     workflow_id,
     start,
     stop,
+    flow_file,
 ) -> Tuple[List[Node], List[Edge]]:
     """Return graph sorted nodes and edges."""
-    config = get_config(workflow_id, opts)
+    config = get_config(workflow_id, opts, flow_file)
     if opts.namespaces:
         nodes, edges = _get_inheritance_nodes_and_edges(config)
     else:
@@ -194,13 +197,8 @@ def _get_inheritance_nodes_and_edges(
     return sorted(nodes), sorted(edges)
 
 
-def get_config(workflow_id: str, opts: 'Values') -> WorkflowConfig:
+def get_config(workflow_id: str, opts: 'Values', flow_file) -> WorkflowConfig:
     """Return a WorkflowConfig object for the provided reg / path."""
-    workflow_id, _, flow_file = parse_id(
-        workflow_id,
-        src=True,
-        constraint='workflows',
-    )
     template_vars = get_template_vars(opts)
     return WorkflowConfig(
         workflow_id, flow_file, opts, template_vars=template_vars
@@ -334,7 +332,7 @@ def open_image(filename):
         img.show()
 
 
-def graph_render(opts, workflow_id, start, stop) -> int:
+def graph_render(opts, workflow_id, start, stop, flow_file) -> int:
     """Render the workflow graph to the specified format.
 
     Graph is rendered to the specified format. The Graphviz "dot" format
@@ -349,6 +347,7 @@ def graph_render(opts, workflow_id, start, stop) -> int:
         workflow_id,
         start,
         stop,
+        flow_file
     )
 
     # format the graph in graphviz-dot format
@@ -382,7 +381,9 @@ def graph_render(opts, workflow_id, start, stop) -> int:
     return 0
 
 
-def graph_reference(opts, workflow_id, start, stop, write=print) -> int:
+def graph_reference(
+    opts, workflow_id, start, stop, flow_file, write=print,
+) -> int:
     """Format the workflow graph using the cylc reference format."""
     # get nodes and edges
     nodes, edges = get_nodes_and_edges(
@@ -390,6 +391,7 @@ def graph_reference(opts, workflow_id, start, stop, write=print) -> int:
         workflow_id,
         start,
         stop,
+        flow_file
     )
     for line in format_cylc_reference(opts, nodes, edges):
         write(line)
@@ -397,13 +399,24 @@ def graph_reference(opts, workflow_id, start, stop, write=print) -> int:
     return 0
 
 
-def graph_diff(opts, workflow_a, workflow_b, start, stop) -> int:
+async def graph_diff(
+    opts, workflow_a, workflow_b, start, stop, flow_file
+) -> int:
     """Difference the workflow graphs using the cylc reference format."""
+
+    workflow_b, _, flow_file_b = await parse_id_async(
+        workflow_b,
+        src=True,
+        constraint='workflows',
+    )
+
     # load graphs
     graph_a: List[str] = []
     graph_b: List[str] = []
-    graph_reference(opts, workflow_a, start, stop, write=graph_a.append),
-    graph_reference(opts, workflow_b, start, stop, write=graph_b.append),
+    graph_reference(
+        opts, workflow_a, start, stop, flow_file, write=graph_a.append),
+    graph_reference(
+        opts, workflow_b, start, stop, flow_file_b, write=graph_b.append),
 
     # compare graphs
     diff_lines = list(
@@ -494,6 +507,9 @@ def get_option_parser() -> COP:
         action='store',
     )
 
+    parser.add_option(
+        *AGAINST_SOURCE_OPTION.args, **AGAINST_SOURCE_OPTION.kwargs)
+
     parser.add_cylc_rose_options()
 
     return parser
@@ -507,20 +523,34 @@ def main(
     start: Optional[str] = None,
     stop: Optional[str] = None
 ) -> None:
+    result = asyncio.run(_main(parser, opts, workflow_id, start, stop))
+    sys.exit(result)
+
+
+async def _main(
+    parser: COP,
+    opts: 'Values',
+    workflow_id: str,
+    start: Optional[str] = None,
+    stop: Optional[str] = None
+) -> int:
     """Implement ``cylc graph``."""
     if opts.grouping and opts.namespaces:
         raise InputError('Cannot combine --group and --namespaces.')
     if opts.cycles and opts.namespaces:
         raise InputError('Cannot combine --cycles and --namespaces.')
 
+    workflow_id, _, flow_file = await parse_id_async(
+        workflow_id,
+        src=True,
+        constraint='workflows',
+    )
+
     if opts.diff:
-        sys.exit(
-            graph_diff(opts, workflow_id, opts.diff, start, stop)
-        )
+        return await graph_diff(
+            opts, workflow_id, opts.diff, start, stop, flow_file)
     if opts.reference:
-        sys.exit(
-            graph_reference(opts, workflow_id, start, stop)
-        )
-    sys.exit(
-        graph_render(opts, workflow_id, start, stop)
-    )
+        return graph_reference(
+            opts, workflow_id, start, stop, flow_file)
+
+    return graph_render(opts, workflow_id, start, stop, flow_file)
diff --git a/cylc/flow/scripts/list.py b/cylc/flow/scripts/list.py
index ed24377e323..c0f1ed18ab7 100755
--- a/cylc/flow/scripts/list.py
+++ b/cylc/flow/scripts/list.py
@@ -30,13 +30,15 @@
   $ cylc graph -n
 """
 
+import asyncio
 import os
 import sys
 from typing import TYPE_CHECKING
 
 from cylc.flow.config import WorkflowConfig
-from cylc.flow.id_cli import parse_id
+from cylc.flow.id_cli import parse_id_async
 from cylc.flow.option_parsers import (
+    AGAINST_SOURCE_OPTION,
     WORKFLOW_ID_OR_PATH_ARG_DOC,
     CylcOptionParser as COP,
     icp_option,
@@ -94,6 +96,9 @@ def get_option_parser():
         "initial cycle point, by default). Use '-p , ' for the default range.",
         metavar="[START],[STOP]", action="store", default=None, dest="prange")
 
+    parser.add_option(
+        *AGAINST_SOURCE_OPTION.args, **AGAINST_SOURCE_OPTION.kwargs)
+
     parser.add_option(icp_option)
 
     parser.add_cylc_rose_options()
@@ -102,7 +107,11 @@ def get_option_parser():
 
 @cli_function(get_option_parser)
 def main(parser: COP, options: 'Values', workflow_id: str) -> None:
-    workflow_id, _, flow_file = parse_id(
+    asyncio.run(_main(parser, options, workflow_id))
+
+
+async def _main(parser: COP, options: 'Values', workflow_id: str) -> None:
+    workflow_id, _, flow_file = await parse_id_async(
         workflow_id,
         src=True,
         constraint='workflows',
diff --git a/cylc/flow/scripts/psutil.py b/cylc/flow/scripts/psutil.py
index 8747acc9a5a..47ff7ec1dbd 100644
--- a/cylc/flow/scripts/psutil.py
+++ b/cylc/flow/scripts/psutil.py
@@ -22,7 +22,7 @@
 the `psutil` on remote platforms.
 
 Exits:
-    0 - If successfull.
+    0 - If successful.
     2 - For errors in extracting results from psutil
     1 - For all other errors.
 """
diff --git a/cylc/flow/scripts/reinstall.py b/cylc/flow/scripts/reinstall.py
index c199c912508..5b0b482109c 100644
--- a/cylc/flow/scripts/reinstall.py
+++ b/cylc/flow/scripts/reinstall.py
@@ -83,6 +83,7 @@
 from cylc.flow.id_cli import parse_id
 from cylc.flow.option_parsers import (
     CylcOptionParser as COP,
+    OptionSettings,
     Options,
     WORKFLOW_ID_ARG_DOC,
 )
@@ -100,6 +101,18 @@
 _input = input  # to enable testing
 
 
+REINSTALL_CYLC_ROSE_OPTIONS = [
+    OptionSettings(
+        ['--clear-rose-install-options'],
+        help="Clear options previously set by cylc-rose.",
+        action='store_true',
+        default=False,
+        dest="clear_rose_install_opts",
+        sources={'reinstall'}
+    )
+]
+
+
 def get_option_parser() -> COP:
     parser = COP(
         __doc__, comms=True, argdoc=[WORKFLOW_ID_ARG_DOC]
@@ -112,14 +125,8 @@ def get_option_parser() -> COP:
     except ImportError:
         pass
     else:
-        parser.add_option(
-            "--clear-rose-install-options",
-            help="Clear options previously set by cylc-rose.",
-            action='store_true',
-            default=False,
-            dest="clear_rose_install_opts"
-        )
-
+        for option in REINSTALL_CYLC_ROSE_OPTIONS:
+            parser.add_option(*option.args, **option.kwargs)
     return parser
 
 
@@ -187,7 +194,6 @@ def reinstall_cli(
 
             display_rose_warning(source)
             display_cylcignore_tip()
-
             # prompt for permission to continue
             while usr not in ['y', 'n']:
                 usr = _input(
@@ -207,14 +213,14 @@ def reinstall_cli(
         reinstall(opts, workflow_id, source, run_dir, dry_run=False)
         print(cparse('<green>Successfully reinstalled.</green>'))
         display_cylc_reload_tip(workflow_id)
+        return True
 
     else:
         # no reinstall
         print(
             cparse('<magenta>Reinstall canceled, no changes made.</magenta>')
         )
-
-    return True
+        return False
 
 
 def reinstall(
diff --git a/cylc/flow/scripts/reload.py b/cylc/flow/scripts/reload.py
index 125e7d40fac..f399cc948cb 100755
--- a/cylc/flow/scripts/reload.py
+++ b/cylc/flow/scripts/reload.py
@@ -104,6 +104,10 @@ async def run(options: 'Values', workflow_id: str) -> None:
 
 @cli_function(get_option_parser)
 def main(parser: COP, options: 'Values', *ids) -> None:
+    reload_cli(options, *ids)
+
+
+def reload_cli(options: 'Values', *ids) -> None:
     call_multi(
         partial(run, options),
         *ids,
diff --git a/cylc/flow/scripts/validate.py b/cylc/flow/scripts/validate.py
index 531dd2d767d..0db83d1bdfa 100755
--- a/cylc/flow/scripts/validate.py
+++ b/cylc/flow/scripts/validate.py
@@ -25,6 +25,7 @@
 use 'cylc view -i,--inline WORKFLOW' for comparison.
 """
 
+import asyncio
 from ansimarkup import parse as cparse
 from copy import deepcopy
 from optparse import Values
@@ -38,9 +39,10 @@
     TriggerExpressionError
 )
 import cylc.flow.flags
-from cylc.flow.id_cli import parse_id
+from cylc.flow.id_cli import parse_id_async
 from cylc.flow.loggingutil import disable_timestamps
 from cylc.flow.option_parsers import (
+    AGAINST_SOURCE_OPTION,
     WORKFLOW_ID_OR_PATH_ARG_DOC,
     CylcOptionParser as COP,
     OptionSettings,
@@ -58,6 +60,8 @@
 VALIDATE_RUN_MODE.sources = {'validate'}
 VALIDATE_ICP_OPTION = deepcopy(ICP_OPTION)
 VALIDATE_ICP_OPTION.sources = {'validate'}
+VALIDATE_AGAINST_SOURCE_OPTION = deepcopy(AGAINST_SOURCE_OPTION)
+VALIDATE_AGAINST_SOURCE_OPTION.sources = {'validate'}
 
 
 VALIDATE_OPTIONS = [
@@ -90,7 +94,8 @@
         sources={'validate'}
     ),
     VALIDATE_RUN_MODE,
-    VALIDATE_ICP_OPTION
+    VALIDATE_ICP_OPTION,
+    VALIDATE_AGAINST_SOURCE_OPTION,
 ]
 
 
@@ -104,10 +109,7 @@ def get_option_parser():
     validate_options = parser.get_cylc_rose_options() + VALIDATE_OPTIONS
 
     for option in validate_options:
-        if isinstance(option, OptionSettings):
-            parser.add_option(*option.args, **option.kwargs)
-        else:
-            parser.add_option(*option['args'], **option['kwargs'])
+        parser.add_option(*option.args, **option.kwargs)
 
     parser.set_defaults(is_validate=True)
 
@@ -127,10 +129,16 @@ def get_option_parser():
 
 @cli_function(get_option_parser)
 def main(parser: COP, options: 'Values', workflow_id: str) -> None:
-    wrapped_main(parser, options, workflow_id)
+    _main(parser, options, workflow_id)
 
 
-def wrapped_main(parser: COP, options: 'Values', workflow_id: str) -> None:
+def _main(parser: COP, options: 'Values', workflow_id: str) -> None:
+    asyncio.run(wrapped_main(parser, options, workflow_id))
+
+
+async def wrapped_main(
+    parser: COP, options: 'Values', workflow_id: str
+) -> None:
     """cylc validate CLI."""
     profiler = Profiler(None, options.profile_mode)
     profiler.start()
@@ -138,7 +146,7 @@ def wrapped_main(parser: COP, options: 'Values', workflow_id: str) -> None:
     if cylc.flow.flags.verbosity < 2:
         disable_timestamps(LOG)
 
-    workflow_id, _, flow_file = parse_id(
+    workflow_id, _, flow_file = await parse_id_async(
         workflow_id,
         src=True,
         constraint='workflows',
diff --git a/cylc/flow/scripts/validate_install_play.py b/cylc/flow/scripts/validate_install_play.py
index 3ed3102dff2..25f6e805b5e 100644
--- a/cylc/flow/scripts/validate_install_play.py
+++ b/cylc/flow/scripts/validate_install_play.py
@@ -30,7 +30,7 @@
 
 from cylc.flow.scripts.validate import (
     VALIDATE_OPTIONS,
-    wrapped_main as validate_main
+    _main as validate_main
 )
 from cylc.flow.scripts.install import (
     INSTALL_OPTIONS, install_cli as cylc_install, get_source_location
@@ -75,7 +75,10 @@ def get_option_parser() -> COP:
         ]
     )
     for option in VIP_OPTIONS:
-        parser.add_option(*option.args, **option.kwargs)
+        # Make a special exception for option against_source which makes
+        # no sense in a VIP context.
+        if option.kwargs.get('dest') != 'against_source':
+            parser.add_option(*option.args, **option.kwargs)
     return parser
 
 
diff --git a/cylc/flow/scripts/validate_reinstall.py b/cylc/flow/scripts/validate_reinstall.py
new file mode 100644
index 00000000000..0d95bbcd6c1
--- /dev/null
+++ b/cylc/flow/scripts/validate_reinstall.py
@@ -0,0 +1,154 @@
+#!/usr/bin/env python3
+
+# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE.
+# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+"""cylc validate-reinstall [OPTIONS] ARGS
+
+Validate, reinstall and apply changes to a workflow.
+
+Validate and reinstall a workflow then either:
+
+* "Reload" the workflow (if it is running),
+* or "play" it (if it is stopped).
+
+This command is equivalent to:
+  $ cylc validate myworkflow --against-source
+  $ cylc reinstall myworkflow
+  # if myworkflow is running:
+  $ cylc reload myworkflow
+  # else:
+  $ cylc play myworkflow
+
+Note:
+  "cylc validate --against-source" checks the code in the workflow source
+  directory against any options (e.g. template variables) which have been set
+  in the installed workflow to ensure the change can be safely applied.
+"""
+
+import sys
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from optparse import Values
+
+from cylc.flow import LOG
+from cylc.flow.exceptions import ServiceFileError
+from cylc.flow.scheduler_cli import PLAY_OPTIONS, scheduler_cli
+from cylc.flow.scripts.validate import (
+    VALIDATE_OPTIONS,
+    _main as cylc_validate
+)
+from cylc.flow.scripts.reinstall import (
+    REINSTALL_CYLC_ROSE_OPTIONS, reinstall_cli as cylc_reinstall
+)
+from cylc.flow.scripts.reload import (
+    reload_cli as cylc_reload
+)
+from cylc.flow.option_parsers import (
+    WORKFLOW_ID_ARG_DOC,
+    CylcOptionParser as COP,
+    combine_options,
+    log_subcommand,
+    cleanup_sysargv
+)
+from cylc.flow.id_cli import parse_id
+from cylc.flow.terminal import cli_function
+from cylc.flow.workflow_files import detect_old_contact_file
+
+
+CYLC_ROSE_OPTIONS = COP.get_cylc_rose_options()
+VR_OPTIONS = combine_options(
+    VALIDATE_OPTIONS,
+    REINSTALL_CYLC_ROSE_OPTIONS,
+    PLAY_OPTIONS,
+    CYLC_ROSE_OPTIONS,
+    modify={'cylc-rose': 'validate, install'}
+)
+
+
+def get_option_parser() -> COP:
+    parser = COP(
+        __doc__,
+        comms=True,
+        jset=True,
+        argdoc=[WORKFLOW_ID_ARG_DOC],
+    )
+    for option in VR_OPTIONS:
+        parser.add_option(*option.args, **option.kwargs)
+    return parser
+
+
+@cli_function(get_option_parser)
+def main(parser: COP, options: 'Values', workflow_id: str):
+    sys.exit(vro_cli(parser, options, workflow_id))
+
+
+def vro_cli(parser: COP, options: 'Values', workflow_id: str):
+    """Run Cylc (re)validate - reinstall - reload in sequence."""
+    # Attempt to work out whether the workflow is running.
+    # We are trying to avoid reinstalling then subsequently being
+    # unable to play or reload because we cannot identify workflow state.
+    workflow_id, *_ = parse_id(
+        workflow_id,
+        constraint='workflows',
+    )
+
+    # Use this interface instead of scan, because it can have an ambiguous
+    # outcome which we want to capture before we install.
+    try:
+        detect_old_contact_file(workflow_id)
+    except ServiceFileError:
+        # Workflow is definately running:
+        workflow_running = True
+    else:
+        # Workflow is definately stopped:
+        workflow_running = False
+
+    # Force on the against_source option:
+    options.against_source = True   # Make validate check against source.
+    log_subcommand('validate --against-source', workflow_id)
+    cylc_validate(parser, options, workflow_id)
+
+    log_subcommand('reinstall', workflow_id)
+    reinstall_ok = cylc_reinstall(options, workflow_id)
+    if not reinstall_ok:
+        LOG.warning(
+            'No changes to source: No reinstall or'
+            f' {"reload" if workflow_running else "play"} required.'
+        )
+        return 1
+
+    # Run reload if workflow is running, else play:
+    if workflow_running:
+        log_subcommand('reload', workflow_id)
+        cylc_reload(options, workflow_id)
+
+    # run play anyway, to resume a paused workflow:
+    else:
+        cleanup_sysargv(
+            'play',
+            workflow_id,
+            options,
+            compound_script_opts=VR_OPTIONS,
+            script_opts=(
+                PLAY_OPTIONS + CYLC_ROSE_OPTIONS
+                + parser.get_std_options()
+            ),
+            source='',  # Intentionally blank
+        )
+        log_subcommand('play', workflow_id)
+        scheduler_cli(options, workflow_id)
diff --git a/cylc/flow/scripts/view.py b/cylc/flow/scripts/view.py
old mode 100755
new mode 100644
index 727bdafe2b4..6c6cf905016
--- a/cylc/flow/scripts/view.py
+++ b/cylc/flow/scripts/view.py
@@ -25,10 +25,12 @@
   configuration (as Cylc would see it).
 """
 
+import asyncio
 from typing import TYPE_CHECKING
 
-from cylc.flow.id_cli import parse_id
+from cylc.flow.id_cli import parse_id_async
 from cylc.flow.option_parsers import (
+    AGAINST_SOURCE_OPTION,
     WORKFLOW_ID_OR_PATH_ARG_DOC,
     CylcOptionParser as COP,
 )
@@ -93,16 +95,19 @@ def get_option_parser():
              "not correspond to those reported by the parser).",
              action="store_true", default=False, dest="cat")
 
+    parser.add_option(
+        *AGAINST_SOURCE_OPTION.args, **AGAINST_SOURCE_OPTION.kwargs)
+
     return parser
 
 
 @cli_function(get_option_parser)
 def main(parser: COP, options: 'Values', workflow_id: str) -> None:
-    _main(options, workflow_id)
+    asyncio.run(_main(options, workflow_id))
 
 
-def _main(options: 'Values', workflow_id: str) -> None:
-    workflow_id, _, flow_file = parse_id(
+async def _main(options: 'Values', workflow_id: str) -> None:
+    workflow_id, _, flow_file = await parse_id_async(
         workflow_id,
         src=True,
         constraint='workflows',
@@ -121,9 +126,8 @@ def _main(options: 'Values', workflow_id: str) -> None:
     }
     for line in read_and_proc(
         flow_file,
-        load_template_vars(
-            options.templatevars, options.templatevars_file
-        ),
-        viewcfg=viewcfg
+        load_template_vars(options.templatevars, options.templatevars_file),
+        viewcfg=viewcfg,
+        opts=options,
     ):
         print(line)
diff --git a/cylc/flow/templatevars.py b/cylc/flow/templatevars.py
index c62e4da5713..97469285d03 100644
--- a/cylc/flow/templatevars.py
+++ b/cylc/flow/templatevars.py
@@ -22,6 +22,21 @@
 from cylc.flow.exceptions import InputError
 
 
+from cylc.flow.rundb import CylcWorkflowDAO
+
+
+def get_template_vars_from_db(run_dir):
+    """Get template vars stored in a workflow run database.
+    """
+    template_vars = {}
+    if (run_dir / 'log/db').exists():
+        dao = CylcWorkflowDAO(str(run_dir / 'log/db'))
+        dao.select_workflow_template_vars(
+            lambda _, row: template_vars.__setitem__(row[0], eval_var(row[1]))
+        )
+    return template_vars
+
+
 def eval_var(var):
     """Wrap ast.literal_eval to provide more helpful error.
 
@@ -65,6 +80,7 @@ def load_template_vars(template_vars=None, template_vars_file=None):
                     continue
                 key, val = line.split("=", 1)
                 res[key.strip()] = eval_var(val.strip())
+
     if template_vars:
         for pair in template_vars:
             key, val = pair.split("=", 1)
@@ -82,6 +98,4 @@ def get_template_vars(options: Values) -> Dict[str, Any]:
     Returns:
         template_vars: Template variables to give to a Cylc config.
     """
-    return load_template_vars(
-        options.templatevars, options.templatevars_file
-    )
+    return load_template_vars(options.templatevars, options.templatevars_file)
diff --git a/cylc/flow/workflow_files.py b/cylc/flow/workflow_files.py
index 9ed83bfe40a..29bbefdab02 100644
--- a/cylc/flow/workflow_files.py
+++ b/cylc/flow/workflow_files.py
@@ -464,7 +464,9 @@ def _is_process_running(
     return cli_format(process['cmdline']) == command
 
 
-def detect_old_contact_file(reg: str, contact_data=None) -> None:
+def detect_old_contact_file(
+    reg: str, contact_data=None
+) -> None:
     """Check if the workflow process is still running.
 
     As a side-effect this should detect and rectify the situation
@@ -480,6 +482,7 @@ def detect_old_contact_file(reg: str, contact_data=None) -> None:
 
     Args:
         reg: workflow name
+        contact_date:
 
     Raises:
         CylcError:
diff --git a/setup.cfg b/setup.cfg
index f6ee306837f..9d8a27c2539 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -200,6 +200,7 @@ cylc.command =
     validate = cylc.flow.scripts.validate:main
     view = cylc.flow.scripts.view:main
     vip = cylc.flow.scripts.validate_install_play:main
+    vr = cylc.flow.scripts.validate_reinstall:main
 # async functions to run within the scheduler main loop
 cylc.main_loop =
     health_check = cylc.flow.main_loop.health_check
diff --git a/tests/functional/cylc-combination-scripts/01-vr-reload.t b/tests/functional/cylc-combination-scripts/01-vr-reload.t
new file mode 100644
index 00000000000..2091b01d111
--- /dev/null
+++ b/tests/functional/cylc-combination-scripts/01-vr-reload.t
@@ -0,0 +1,57 @@
+#!/usr/bin/env bash
+# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE.
+# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+#------------------------------------------------------------------------------
+# Test `cylc vr` (Validate Reinstall)
+# In this case the target workflow is running so cylc reload is run.
+
+. "$(dirname "$0")/test_header"
+set_test_number 7
+
+
+# Setup (Must be a running workflow, note the unusual absence of --no-detach)
+WORKFLOW_NAME="cylctb-x$(< /dev/urandom tr -dc _A-Z-a-z-0-9 | head -c6)"
+cp "${TEST_SOURCE_DIR}/vr_workflow/flow.cylc" .
+run_ok "setup (vip)" \
+    cylc vip --debug \
+    --workflow-name "${WORKFLOW_NAME}" \
+    --no-run-name
+export WORKFLOW_RUN_DIR="${RUN_DIR}/${WORKFLOW_NAME}"
+poll_workflow_running
+
+
+# It validates and reloads:
+
+sed -i 's@P1Y@P5Y@' flow.cylc
+run_ok "${TEST_NAME_BASE}-runs" cylc vr "${WORKFLOW_NAME}"
+
+# Grep for VR reporting revalidation, reinstallation and reloading
+grep "\$" "${TEST_NAME_BASE}-runs.stdout" > VIPOUT.txt
+named_grep_ok "${TEST_NAME_BASE}-it-validated" "$ cylc validate --against-source" "VIPOUT.txt"
+named_grep_ok "${TEST_NAME_BASE}-it-installed" "$ cylc reinstall" "VIPOUT.txt"
+named_grep_ok "${TEST_NAME_BASE}-it-reloaded" "$ cylc reload" "VIPOUT.txt"
+
+cylc play "${WORKFLOW_NAME}"
+
+named_grep_ok "${TEST_NAME_BASE}-it-logged-reload" \
+    "Reloading the workflow definition" \
+    "${WORKFLOW_RUN_DIR}/log/scheduler/log"
+
+# Clean Up.
+run_ok "teardown (stop workflow)" cylc stop "${WORKFLOW_NAME}" --now --now
+purge
+exit 0
diff --git a/tests/functional/cylc-combination-scripts/02-vr-restart.t b/tests/functional/cylc-combination-scripts/02-vr-restart.t
new file mode 100644
index 00000000000..2a746a4c769
--- /dev/null
+++ b/tests/functional/cylc-combination-scripts/02-vr-restart.t
@@ -0,0 +1,52 @@
+#!/usr/bin/env bash
+# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE.
+# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+#------------------------------------------------------------------------------
+# Test `cylc vr` (Validate Reinstall restart)
+# In this case the target workflow is stopped so cylc play is run.
+
+
+. "$(dirname "$0")/test_header"
+set_test_number 6
+
+# Setup
+WORKFLOW_NAME="cylctb-x$(< /dev/urandom tr -dc _A-Z-a-z-0-9 | head -c6)"
+cp "${TEST_SOURCE_DIR}/vr_workflow/flow.cylc" .
+run_ok "setup (install)" \
+    cylc install \
+    --workflow-name "${WORKFLOW_NAME}" \
+    --no-run-name
+
+export WORKFLOW_RUN_DIR="${RUN_DIR}/${WORKFLOW_NAME}"
+
+# It validates and restarts:
+
+# Change source workflow and run vr:
+sed -i 's@P1Y@P5Y@' flow.cylc
+run_ok "${TEST_NAME_BASE}-runs" cylc vr "${WORKFLOW_NAME}"
+
+# Grep for vr reporting revalidation, reinstallation and playing the workflow.
+grep "\$" "${TEST_NAME_BASE}-runs.stdout" > VIPOUT.txt
+named_grep_ok "${TEST_NAME_BASE}-it-revalidated" "$ cylc validate --against-source" "VIPOUT.txt"
+named_grep_ok "${TEST_NAME_BASE}-it-installed" "$ cylc reinstall" "VIPOUT.txt"
+named_grep_ok "${TEST_NAME_BASE}-it-played" "$ cylc play" "VIPOUT.txt"
+
+
+# Clean Up.
+run_ok "teardown (stop workflow)" cylc stop "${WORKFLOW_NAME}" --now --now
+purge
+exit 0
diff --git a/tests/functional/cylc-combination-scripts/04-vr-fail-validate.t b/tests/functional/cylc-combination-scripts/04-vr-fail-validate.t
new file mode 100644
index 00000000000..ef819a32781
--- /dev/null
+++ b/tests/functional/cylc-combination-scripts/04-vr-fail-validate.t
@@ -0,0 +1,53 @@
+#!/usr/bin/env bash
+# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE.
+# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+#------------------------------------------------------------------------------
+# Test `cylc vr` (Validate Reinstall restart)
+# Changes to the source cause VR to bail on validation.
+
+. "$(dirname "$0")/test_header"
+set_test_number 5
+
+# Setup
+WORKFLOW_NAME="cylctb-x$(< /dev/urandom tr -dc _A-Z-a-z-0-9 | head -c6)"
+cp "${TEST_SOURCE_DIR}/vr_workflow/flow.cylc" .
+run_ok "setup (vip)" \
+    cylc vip --debug \
+    --workflow-name "${WORKFLOW_NAME}" \
+    --no-run-name
+
+
+# Change source workflow and run vr:
+
+# Cut the runtime section out of the source flow.
+head -n 5 > tmp < flow.cylc
+cat tmp > flow.cylc
+
+TEST_NAME="${TEST_NAME_BASE}"
+run_fail "${TEST_NAME}" cylc vr "${WORKFLOW_NAME}"
+
+# Grep for reporting of revalidation, reinstallation, reloading and playing:
+named_grep_ok "${TEST_NAME_BASE}-it-tried" \
+    "$ cylc validate --against-source" "${TEST_NAME}.stdout"
+named_grep_ok "${TEST_NAME_BASE}-it-failed" \
+    "WorkflowConfigError" "${TEST_NAME}.stderr"
+
+
+# Clean Up:
+run_ok "teardown (stop workflow)" cylc stop "${WORKFLOW_NAME}" --now --now
+purge
+exit 0
diff --git a/tests/functional/cylc-combination-scripts/05-vr-fail-is-running.t b/tests/functional/cylc-combination-scripts/05-vr-fail-is-running.t
new file mode 100644
index 00000000000..876c494d4f3
--- /dev/null
+++ b/tests/functional/cylc-combination-scripts/05-vr-fail-is-running.t
@@ -0,0 +1,60 @@
+#!/usr/bin/env bash
+# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE.
+# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+#------------------------------------------------------------------------------
+# Test `cylc vr` (Validate Reinstall)
+# In this case the target workflow is in an abiguous state: We cannot tell
+# Whether it's running, paused or stopped. Cylc VR should validate before
+# reinstall:
+
+. "$(dirname "$0")/test_header"
+set_test_number 4
+
+create_test_global_config "" """
+[scheduler]
+    [[main loop]]
+        plugins = reset bad hosts
+"""
+
+# Setup
+WORKFLOW_NAME="cylctb-x$(< /dev/urandom tr -dc _A-Z-a-z-0-9 | head -c6)"
+cp "${TEST_SOURCE_DIR}/vr_workflow/flow.cylc" .
+run_ok "setup (vip)" \
+    cylc vip --debug \
+    --workflow-name "${WORKFLOW_NAME}" \
+    --no-run-name
+
+
+# Get the workflow into an unreachable state
+CONTACTFILE="${RUN_DIR}/${WORKFLOW_NAME}/.service/contact"
+poll test -e "${CONTACTFILE}"
+
+sed -i 's@CYLC_WORKFLOW_HOST=.*@CYLC_WORKFLOW_HOST=elephantshrew@' "${CONTACTFILE}"
+
+
+# It can't figure out whether the workflow is running:
+
+# Change source workflow and run vr:
+run_fail "${TEST_NAME_BASE}-runs" cylc vr "${WORKFLOW_NAME}"
+
+grep_ok "on elephantshrew." "${TEST_NAME_BASE}-runs.stderr"
+
+# Clean Up:
+sed -i "s@CYLC_WORKFLOW_HOST=elephantshrew@CYLC_WORKFLOW_HOST=$HOSTNAME@" "${CONTACTFILE}"
+run_ok "teardown (stop workflow)" cylc stop "${WORKFLOW_NAME}" --now --now
+purge
+exit 0
diff --git a/tests/functional/cylc-combination-scripts/vr_workflow/flow.cylc b/tests/functional/cylc-combination-scripts/vr_workflow/flow.cylc
new file mode 100644
index 00000000000..6014d4dfd36
--- /dev/null
+++ b/tests/functional/cylc-combination-scripts/vr_workflow/flow.cylc
@@ -0,0 +1,8 @@
+[scheduling]
+    initial cycle point = 1500
+    [[graph]]
+        P1Y = foo
+
+[runtime]
+    [[foo]]
+        script = cylc pause ${WORKFLOW_ID}
diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py
index 02ef139ae69..65fdab9713a 100644
--- a/tests/integration/conftest.py
+++ b/tests/integration/conftest.py
@@ -20,18 +20,25 @@
 from pathlib import Path
 import pytest
 from shutil import rmtree
-from typing import List, TYPE_CHECKING, Set, Tuple
+from typing import List, TYPE_CHECKING, Set, Tuple, Union
 
 from cylc.flow.config import WorkflowConfig
+from cylc.flow.option_parsers import Options
 from cylc.flow.network.client import WorkflowRuntimeClient
 from cylc.flow.pathutil import get_cylc_run_dir
 from cylc.flow.rundb import CylcWorkflowDAO
 from cylc.flow.scripts.validate import ValidateOptions
+from cylc.flow.scripts.install import (
+    install as cylc_install,
+    get_option_parser as install_gop
+)
 from cylc.flow.wallclock import get_current_time_string
+from cylc.flow.workflow_files import infer_latest_run_from_id
 
 from .utils import _rm_if_empty
 from .utils.flow_tools import (
     _make_flow,
+    _make_src_flow,
     _make_scheduler,
     _run_flow,
     _start_flow,
@@ -42,6 +49,9 @@
     from cylc.flow.task_proxy import TaskProxy
 
 
+InstallOpts = Options(install_gop())
+
+
 @pytest.hookimpl(tryfirst=True, hookwrapper=True)
 def pytest_runtest_makereport(item, call):
     """Expose the result of tests to their fixtures.
@@ -335,7 +345,8 @@ def validate(run_dir):
         reg - The flow to validate
         kwargs - Arguments to pass to ValidateOptions
     """
-    def _validate(reg: str, **kwargs) -> WorkflowConfig:
+    def _validate(reg: Union[str, Path], **kwargs) -> WorkflowConfig:
+        reg = str(reg)
         return WorkflowConfig(
             reg,
             str(Path(run_dir, reg, 'flow.cylc')),
@@ -404,3 +415,61 @@ def run_job_cmd(
         return polled_tasks
 
     return _disable_polling
+
+
+@pytest.fixture(scope='module')
+def mod_workflow_source(mod_flow, tmp_path_factory):
+    """Create a workflow source directory.
+
+    Args:
+        cfg: Can be passed a config dictionary.
+
+    Yields:
+        Path to source directory.
+    """
+    def _inner(cfg):
+        src_dir = _make_src_flow(tmp_path_factory.getbasetemp(), cfg)
+        return src_dir
+    yield _inner
+
+
+@pytest.fixture
+def workflow_source(mod_flow, tmp_path):
+    """Create a workflow source directory.
+
+    Args:
+        cfg: Can be passed a config dictionary.
+
+    Yields:
+        Path to source directory.
+    """
+    def _inner(cfg):
+        src_dir = _make_src_flow(tmp_path, cfg)
+        return src_dir
+    yield _inner
+
+
+@pytest.fixture
+def install(test_dir, run_dir):
+    """Install a workflow from source
+
+    Args:
+        (Actually args for _inner, but what the fixture appears to take to
+        the user)
+        source: Directory containing the source.
+        **kwargs: Options for cylc install.
+
+    Returns:
+        Workflow id, including run directory.
+    """
+    def _inner(source, **kwargs):
+        opts = InstallOpts(**kwargs)
+        # Note we append the source.name to the string rather than creating
+        # a subfolder because the extra layer of directories would exceed
+        # Cylc install's default limit.
+        opts.workflow_name = (
+            f'{str(test_dir.relative_to(run_dir))}.{source.name}')
+        workflow_id = cylc_install(opts, str(source))
+        workflow_id = infer_latest_run_from_id(workflow_id)
+        return workflow_id
+    yield _inner
diff --git a/tests/integration/scripts/test_validate_integration.py b/tests/integration/scripts/test_validate_integration.py
new file mode 100644
index 00000000000..e271726dafb
--- /dev/null
+++ b/tests/integration/scripts/test_validate_integration.py
@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+
+# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE.
+# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Integration tests for Cylc Validate CLI script."""
+
+from cylc.flow.parsec.exceptions import IllegalItemError
+import pytest
+from cylc.flow.parsec.exceptions import Jinja2Error
+
+
+async def test_validate_against_source_checks_source(
+    capsys, validate, workflow_source, install, one_conf
+):
+    """Validation fails if validating against source with broken config.
+    """
+    src_dir = workflow_source(one_conf)
+    workflow_id = install(src_dir)
+
+    # Check that the original installation validates OK:
+    validate(workflow_id, against_source=True)
+
+    # Break the source config:
+    with open(src_dir / 'flow.cylc', 'a') as handle:
+        handle.write('\n[runtime]\n[[foo]]\nAgrajag = bowl of petunias')
+
+    # # Check that Validate now fails:
+    with pytest.raises(IllegalItemError, match='Agrajag'):
+        validate(workflow_id, against_source=True)
+
+
+async def test_validate_against_source_gets_old_tvars(
+    workflow_source, capsys, validate, scheduler, run, install, run_dir
+):
+    """Validation will retrieve template variables from a previously played
+    workflow.
+    """
+    src_dir = workflow_source({
+        '#!jinja2': None,
+        'scheduler': {
+            'allow implicit tasks': True
+        },
+        'scheduling': {
+            'initial cycle point': '1854',
+            'graph': {
+                'P1Y': 'foo'
+            },
+        },
+        'runtime': {
+            'foo': {
+                'script': 'cylc pause ${CYLC_WORKFLOW_ID}'
+            }
+        }
+    })
+
+    wf_id = install(src_dir)
+    installed_dir = run_dir / wf_id
+
+    # Check that the original installation validates OK:
+    validate(installed_dir)
+
+    # # Start a scheduler with tvars option:
+    schd = scheduler(
+        wf_id,
+        templatevars=['FOO="foo"']
+    )
+    async with run(schd):
+        pass
+
+    # Replace foo in graph with {{FOO}} and check that this still
+    # Validates (using db value for FOO):
+    flow_file = (installed_dir / 'flow.cylc')
+    flow_file.write_text(
+        flow_file.read_text().replace('P1Y = foo', 'P1Y = {{FOO}}'))
+    validate(wf_id, against_source=True)
+
+    # Check that the source will not validate alone:
+    flow_file = (src_dir / 'flow.cylc')
+    flow_file.write_text(
+        flow_file.read_text().replace('P1Y = foo', 'P1Y = {{FOO}}'))
+    with pytest.raises(Jinja2Error):
+        validate(src_dir)
diff --git a/tests/integration/test_get_old_tvars.py b/tests/integration/test_get_old_tvars.py
new file mode 100644
index 00000000000..cd1ee2f118d
--- /dev/null
+++ b/tests/integration/test_get_old_tvars.py
@@ -0,0 +1,92 @@
+# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE.
+# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import pytest
+from pytest import param
+
+from cylc.flow.option_parsers import Options
+
+from cylc.flow.scripts.validate import (
+    wrapped_main as validate,
+    get_option_parser as validate_gop
+)
+from cylc.flow.scripts.view import (
+    _main as view,
+    get_option_parser as view_gop
+)
+from cylc.flow.scripts.graph import (
+    _main as graph,
+    get_option_parser as graph_gop
+)
+from cylc.flow.scripts.config import (
+    _main as config,
+    get_option_parser as config_gop
+)
+from cylc.flow.scripts.list import (
+    _main as cylclist,
+    get_option_parser as list_gop
+)
+
+
+@pytest.fixture(scope='module')
+def _setup(mod_scheduler, mod_flow):
+    """Provide an installed flow with a database to try assorted
+    simple Cylc scripts against.
+    """
+    conf = {
+        '#!jinja2': '',
+        'scheduler': {
+            'allow implicit tasks': True
+        },
+        'scheduling': {
+            'graph': {
+                'R1': r'{{FOO}}'
+            }
+        }
+    }
+    schd = mod_scheduler(mod_flow(conf), templatevars=['FOO="bar"'])
+    yield schd
+
+
+@pytest.mark.parametrize(
+    'function, parser, expect',
+    (
+        param(validate, validate_gop, 'Valid for', id="validate"),
+        param(view, view_gop, 'FOO', id="view"),
+        param(graph, graph_gop, '1/bar', id='graph'),
+        param(config, config_gop, 'R1 = bar', id='config'),
+        param(cylclist, list_gop, 'bar', id='list')
+    )
+)
+async def test_validate_with_old_tvars(
+    _setup, mod_start, capsys, function, parser, expect,
+):
+    """It (A Cylc CLI Command - see parameters) can
+    get template vars stored in db.
+
+    Else the jinja2 in the config would cause these tasks to fail.
+    """
+    parser = parser()
+    opts = Options(parser)()
+    if function == graph:
+        opts.reference = True
+
+    async with mod_start(_setup):
+        if function == view:
+            await function(opts, _setup.workflow_name)
+        else:
+            await function(parser, opts, _setup.workflow_name)
+        assert expect in capsys.readouterr().out
diff --git a/tests/integration/test_reinstall.py b/tests/integration/test_reinstall.py
index a0d2cdbc12a..02be406bc53 100644
--- a/tests/integration/test_reinstall.py
+++ b/tests/integration/test_reinstall.py
@@ -155,7 +155,7 @@ def test_interactive(
         'cylc.flow.scripts.reinstall._input',
         lambda x: 'n'
     )
-    assert reinstall_cli(opts=ReInstallOptions(), args=one_run.id)
+    assert reinstall_cli(opts=ReInstallOptions(), args=one_run.id) is False
 
     # only one rsync call should have been made (i.e. the --dry-run)
     assert [call[1].get('dry_run') for call in reinstall_calls] == [True]
@@ -257,7 +257,7 @@ def test_keyboard_interrupt(
     one_run,
     interactive,
     monkeypatch,
-    capsys,
+    capsys
 ):
     """It should handle a KeyboardInterrupt during dry-run elegantly.
 
@@ -274,7 +274,6 @@ def raise_keyboard_interrupt():
     )
 
     reinstall_cli(opts=ReInstallOptions(), args=one_run.id)
-
     assert 'Reinstall canceled, no changes made' in capsys.readouterr().out
 
 
diff --git a/tests/integration/utils/flow_tools.py b/tests/integration/utils/flow_tools.py
index dcd683023bb..2a172a09140 100644
--- a/tests/integration/utils/flow_tools.py
+++ b/tests/integration/utils/flow_tools.py
@@ -39,11 +39,22 @@
 from .flow_writer import flow_config_str
 
 
+def _make_src_flow(src_path, conf):
+    """Construct a workflow on the filesystem"""
+    flow_src_dir = (src_path / str(uuid1()))
+    flow_src_dir.mkdir(parents=True, exist_ok=True)
+    if isinstance(conf, dict):
+        conf = flow_config_str(conf)
+    with open((flow_src_dir / WorkflowFiles.FLOW_FILE), 'w+') as flow_file:
+        flow_file.write(conf)
+    return flow_src_dir
+
+
 def _make_flow(
     cylc_run_dir: Union[Path, str],
     test_dir: Path,
     conf: Union[dict, str],
-    name: Optional[str] = None
+    name: Optional[str] = None,
 ) -> str:
     """Construct a workflow on the filesystem."""
     if name is None:
diff --git a/tests/integration/utils/flow_writer.py b/tests/integration/utils/flow_writer.py
index 68bc3acdc28..c1e21449593 100644
--- a/tests/integration/utils/flow_writer.py
+++ b/tests/integration/utils/flow_writer.py
@@ -63,7 +63,9 @@ def _write_conf(conf: dict, level: int) -> List[str]:
     ret = []
     for key, value in conf.items():
         # write out settings first
-        if not isinstance(value, dict):
+        if key.lower() == '#!jinja2':
+            ret.append('#!jinja2')
+        elif not isinstance(value, dict):
             ret.extend(
                 _write_setting(key, value, level + 1)
             )
diff --git a/tests/integration/utils/test_flow_writer.py b/tests/integration/utils/test_flow_writer.py
index a4cf55d36c4..0913a22bf37 100644
--- a/tests/integration/utils/test_flow_writer.py
+++ b/tests/integration/utils/test_flow_writer.py
@@ -97,6 +97,7 @@ def test_flow_config_str():
     """It should write out entire cylc configuration files."""
     assert flow_config_str(
         {
+            '#!JiNjA2': None,
             'foo': {
                 'bar': {
                     'pub': 'beer'
@@ -106,6 +107,7 @@ def test_flow_config_str():
             'qux': 'asdf'
         }
     ) == dedent('''
+        #!jinja2
         qux = asdf
         [foo]
             baz = 42
diff --git a/tests/unit/parsec/test_fileparse.py b/tests/unit/parsec/test_fileparse.py
index 34fad5ff30d..8a972ae9b5c 100644
--- a/tests/unit/parsec/test_fileparse.py
+++ b/tests/unit/parsec/test_fileparse.py
@@ -16,15 +16,22 @@
 
 import tempfile
 
+import os
 import pytest
+from pytest import param
+import sqlite3
+from types import SimpleNamespace
 
 from cylc.flow.parsec.exceptions import (
     FileParseError,
     IncludeFileNotFoundError,
     Jinja2Error,
+    ParsecError,
 )
 from cylc.flow.parsec.OrderedDict import OrderedDictWithDefaults
 from cylc.flow.parsec.fileparse import (
+    _prepend_old_templatevars,
+    _get_fpath_for_source,
     addict,
     addsect,
     multiline,
@@ -599,3 +606,95 @@ def test_unclosed_multiline():
 def test_merge_template_vars(caplog, expect, native_tvars, plugin_result, log):
     assert merge_template_vars(native_tvars, plugin_result) == expect
     assert [r.msg for r in caplog.records] == log
+
+
+@pytest.fixture
+def _mock_old_template_vars_db(tmp_path):
+    def _inner(create_srclink=True):
+        # Create a fake workflow dir:
+        (tmp_path / 'flow.cylc').touch()
+        (tmp_path / 'log').mkdir()
+        db_path = tmp_path / 'log/db'
+        db_path.touch()
+
+        # Set up a fake workflow database:
+        conn = sqlite3.connect(str(db_path))
+        conn.execute(
+            "CREATE TABLE workflow_template_vars"
+            "(key TEXT, value TEXT, PRIMARY KEY(key)) ;"
+        )
+        conn.execute(
+            "INSERT INTO workflow_template_vars VALUES"
+            "    ('Marius', '\"Consul\"')"
+        )
+        conn.commit()
+        conn.close()
+
+        # Simulate being an installed rundir by creating a sourcelink:
+        if create_srclink:
+            src = (tmp_path / 'src')
+            src.mkdir(exist_ok=True)
+            link = tmp_path.parent / '_cylc-install/source'
+            link.parent.mkdir(exist_ok=True)
+            try:
+                os.symlink(src, link)
+            except FileExistsError:
+                # We don't mind the link persisting.
+                pass
+        return tmp_path / 'flow.cylc'
+    yield _inner
+
+
+@pytest.mark.parametrize(
+    'expect, tvars',
+    [
+        param(
+            {'Marius': 'Consul', 'Julius': 'Pontifex'}, {'Julius': 'Pontifex'},
+            id='It adds a new key at the end'
+        ),
+        param(
+            {'Marius': 'Tribune'}, {'Marius': 'Tribune'},
+            id='It overrides an existing key'
+        ),
+    ]
+)
+def test__prepend_old_templatevars2(_mock_old_template_vars_db, expect, tvars):
+    # Create a target for a source symlink
+    result = _prepend_old_templatevars(
+        _mock_old_template_vars_db(), tvars)
+    assert result == expect
+
+
+def test_get_fpath_for_source(tmp_path):
+    # Create rundir and srcdir:
+    srcdir = tmp_path / 'source'
+    rundir = tmp_path / 'run'
+    srcdir.mkdir()
+    rundir.mkdir()
+    (rundir / 'flow.cylc').touch()
+    (srcdir / 'flow.cylc').touch()
+
+    # Mock Options object:
+    opts = SimpleNamespace()
+
+    # It raises an error if source is not linked:
+    with pytest.raises(
+        ParsecError, match=f'Cannot validate {rundir} against source:'
+    ):
+        opts.against_source = True
+        _get_fpath_for_source(rundir / 'flow.cylc', opts)
+
+    # Create symlinks:
+    link = rundir / '_cylc-install/source'
+    link.parent.mkdir(exist_ok=True)
+    os.symlink(srcdir, link)
+
+    # It does nothing if opts.against_source is False:
+    opts.against_source = False
+    assert _get_fpath_for_source(
+        rundir / 'flow.cylc', opts) == rundir / 'flow.cylc'
+
+    # It gets source dir if opts.against_source is True:
+    opts.against_source = True
+    assert _get_fpath_for_source(
+        rundir / 'flow.cylc', opts) == str(srcdir / 'flow.cylc')
diff --git a/tests/unit/scripts/test_graph.py b/tests/unit/scripts/test_graph.py
index aa64002fca7..f20c6377d1c 100644
--- a/tests/unit/scripts/test_graph.py
+++ b/tests/unit/scripts/test_graph.py
@@ -89,7 +89,7 @@ def _get_parents_lists(*args, **kwargs):
 
     monkeypatch.setattr(
         'cylc.flow.scripts.graph.get_config',
-        lambda x, y: config
+        lambda x, y, z: config
     )
 
 
@@ -295,11 +295,11 @@ def test_null(null_config):
         grouping=False,
         show_suicide=False
     )
-    assert get_nodes_and_edges(opts, None, 1, 2) == ([], [])
+    assert get_nodes_and_edges(opts, None, 1, 2, '') == ([], [])
 
     opts = SimpleNamespace(
         namespaces=True,
         grouping=False,
         show_suicide=False
     )
-    assert get_nodes_and_edges(opts, None, 1, 2) == ([], [])
+    assert get_nodes_and_edges(opts, None, 1, 2, '')  == ([], [])
diff --git a/tests/unit/scripts/test_trigger.py b/tests/unit/scripts/test_trigger.py
index 6e4476eea36..d464bda0a34 100644
--- a/tests/unit/scripts/test_trigger.py
+++ b/tests/unit/scripts/test_trigger.py
@@ -108,7 +108,6 @@
                 "Multiple flow options must all be integer valued"
             )
         ),
- 
     ]
 )
 def test_validate(
diff --git a/tests/unit/scripts/test_validate.py b/tests/unit/scripts/test_validate.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/tests/unit/test_option_parsers.py b/tests/unit/test_option_parsers.py
index 004e6508aa1..332851d34c4 100644
--- a/tests/unit/test_option_parsers.py
+++ b/tests/unit/test_option_parsers.py
@@ -146,7 +146,7 @@ def test_Options_std_opts():
             [{ARGS: ['-f', '--foo'], KWARGS: {}, SOURCES: {'cook'}}],
             [{
                 ARGS: ['-f', '--foo'],
-                KWARGS: {'help': 'not identical'},
+                KWARGS: {'help': 'not identical', 'dest': 'foobius'},
                 SOURCES: {'bake'},
                 USEIF: ''
             }],
diff --git a/tests/unit/test_pathutil.py b/tests/unit/test_pathutil.py
index 67a779e6b7e..d42b26a2dbe 100644
--- a/tests/unit/test_pathutil.py
+++ b/tests/unit/test_pathutil.py
@@ -529,7 +529,8 @@ def test_parse_rm_dirs__bad(dirs: List[str], err_msg: str):
             1000, ['run20', 'run400', 'run999'], False,
             id='1000th run (from filenames)'
         ),
-        param(6, ['run1', 'run5'], False,
+        param(
+            6, ['run1', 'run5'], False,
             id='Non-sequential (from filenames)'),
         param(2, ['run1'], True, id='2nd run (from symlink)'),
         param(100, ['run1', 'run99'], True, id='100th run (from symlink)'),
@@ -549,7 +550,8 @@ def test_get_next_rundir_number(tmp_path, expect, files, runN):
     (
         param('my_workflow1', 'my_workflow1', False, id='--no-run-name'),
         param('my_workflow2', 'my_workflow2/run22', False, id='installed'),
-        param('my_workflow3', 'my_workflow3/foo', False, id='--run-name="foo"'),
+        param(
+            'my_workflow3', 'my_workflow3/foo', False, id='--run-name="foo"'),
         param('my_workflow4', 'my_workflow4', True, id='not installed'),
     )
 )
diff --git a/tests/unit/test_templatevars.py b/tests/unit/test_templatevars.py
index f2f2182e552..96821098499 100644
--- a/tests/unit/test_templatevars.py
+++ b/tests/unit/test_templatevars.py
@@ -15,13 +15,19 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import pytest
+import sqlite3
 import tempfile
 import unittest
 
 from types import SimpleNamespace
 
+
 from cylc.flow.exceptions import PluginError
-from cylc.flow.templatevars import get_template_vars, load_template_vars
+from cylc.flow.templatevars import (
+    get_template_vars_from_db,
+    get_template_vars,
+    load_template_vars
+)
 
 
 class TestTemplatevars(unittest.TestCase):
@@ -102,3 +108,48 @@ def test_load_template_vars_from_string_and_file_2(self):
             'none': None
         }
         self.assertEqual(expected, load_template_vars(template_vars=pairs))
+
+
+@pytest.fixture(scope='module')
+def _setup_db(tmp_path_factory):
+    tmp_path = tmp_path_factory.mktemp('test_get_old_tvars')
+    logfolder = tmp_path / "log/"
+    logfolder.mkdir()
+    db_path = logfolder / 'db'
+    conn = sqlite3.connect(db_path)
+    conn.execute(
+        r'''
+            CREATE TABLE workflow_template_vars (
+                key,
+                value
+            )
+        '''
+    )
+    conn.execute(
+        r'''
+            INSERT INTO workflow_template_vars
+            VALUES
+                ("FOO", "42"),
+                ("BAR", "'hello world'"),
+                ("BAZ", "'foo', 'bar', 48"),
+                ("QUX", "['foo', 'bar', 21]")
+        '''
+    )
+    conn.commit()
+    conn.close()
+    yield get_template_vars_from_db(tmp_path)
+
+
+@pytest.mark.parametrize(
+    'key, expect',
+    (
+        ('FOO', 42),
+        ('BAR', 'hello world'),
+        ('BAZ', ('foo', 'bar', 48)),
+        ('QUX', ['foo', 'bar', 21])
+    )
+)
+def test_get_old_tvars(key, expect, _setup_db):
+    """It can extract a variety of items from a workflow database.
+    """
+    assert _setup_db[key] == expect