diff --git a/scripts/dispatcher.py b/scripts/dispatcher.py
index 4cb1abdea6..45dd61b264 100644
--- a/scripts/dispatcher.py
+++ b/scripts/dispatcher.py
@@ -1,35 +1,62 @@
-"""
-usage: teuthology-dispatcher --help
-       teuthology-dispatcher --supervisor [-v] --bin-path BIN_PATH --job-config COFNFIG --archive-dir DIR
-       teuthology-dispatcher [-v] [--archive-dir DIR] [--exit-on-empty-queue] --log-dir LOG_DIR --tube TUBE
-
-Start a dispatcher for the specified tube. Grab jobs from a beanstalk
-queue and run the teuthology tests they describe as subprocesses. The
-subprocess invoked is a teuthology-dispatcher command run in supervisor
-mode.
-
-Supervisor mode: Supervise the job run described by its config. Reimage
-target machines and invoke teuthology command. Unlock the target machines
-at the end of the run.
-
-standard arguments:
-  -h, --help                     show this help message and exit
-  -v, --verbose                  be more verbose
-  -t, --tube TUBE                which beanstalk tube to read jobs from
-  -l, --log-dir LOG_DIR          path in which to store logs
-  -a DIR, --archive-dir DIR      path to archive results in
-  --supervisor                   run dispatcher in job supervisor mode
-  --bin-path BIN_PATH            teuthology bin path
-  --job-config CONFIG            file descriptor of job's config file
-  --exit-on-empty-queue          if the queue is empty, exit
-"""
-
-import docopt
+import argparse
 import sys
 
-import teuthology.dispatcher
+import teuthology.dispatcher.supervisor
+
+from .supervisor import parse_args as parse_supervisor_args
+
+
+def parse_args(argv):
+    parser = argparse.ArgumentParser(
+        description="Start a dispatcher for the specified tube. Grab jobs from a beanstalk queue and run the teuthology tests they describe as subprocesses. The subprocess invoked is teuthology-supervisor."
+    )
+    parser.add_argument(
+        "-v",
+        "--verbose",
+        action="store_true",
+        help="be more verbose",
+    )
+    parser.add_argument(
+        "-a",
+        "--archive-dir",
+        type=str,
+        help="path to archive results in",
+    )
+    parser.add_argument(
+        "-t",
+        "--tube",
+        type=str,
+        help="which beanstalk tube to read jobs from",
+        required=True,
+    )
+    parser.add_argument(
+        "-l",
+        "--log-dir",
+        type=str,
+        help="path in which to store the dispatcher log",
+        required=True,
+    )
+    parser.add_argument(
+        "--exit-on-empty-queue",
+        action="store_true",
+        help="if the queue is empty, exit",
+    )
+    return parser.parse_args(argv)
 
 
 def main():
-    args = docopt.docopt(__doc__)
-    sys.exit(teuthology.dispatcher.main(args))
+    if "--supervisor" in sys.argv:
+        # This is for transitional compatibility, so the old dispatcher can
+        # invoke the new supervisor. Once old dispatchers are phased out,
+        # this block can be as well.
+        sys.argv.remove("--supervisor")
+        sys.argv[0] = "teuthology-supervisor"
+        sys.exit(teuthology.dispatcher.supervisor.main(
+            parse_supervisor_args(sys.argv[1:])
+        ))
+    else:
+        sys.exit(teuthology.dispatcher.main(parse_args(sys.argv[1:])))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/scripts/supervisor.py b/scripts/supervisor.py
new file mode 100644
index 0000000000..7450473eb1
--- /dev/null
+++ b/scripts/supervisor.py
@@ -0,0 +1,44 @@
+import argparse
+import sys
+
+import teuthology.dispatcher.supervisor
+
+
+def parse_args(argv):
+    parser = argparse.ArgumentParser(
+        description="Supervise and run a teuthology job; normally only run by the dispatcher",
+    )
+    parser.add_argument(
+        "-v",
+        "--verbose",
+        action="store_true",
+        help="be more verbose",
+    )
+    parser.add_argument(
+        "-a",
+        "--archive-dir",
+        type=str,
+        help="path in which to store the job's logfiles",
+        required=True,
+    )
+    parser.add_argument(
+        "--bin-path",
+        type=str,
+        help="teuthology bin path",
+        required=True,
+    )
+    parser.add_argument(
+        "--job-config",
+        type=str,
+        help="file descriptor of job's config file",
+        required=True,
+    )
+    return parser.parse_args(argv)
+
+
+def main():
+    sys.exit(teuthology.dispatcher.supervisor.main(parse_args(sys.argv[1:])))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/scripts/test/test_dispatcher_.py b/scripts/test/test_dispatcher_.py
new file mode 100644
index 0000000000..4d201aae5d
--- /dev/null
+++ b/scripts/test/test_dispatcher_.py
@@ -0,0 +1,5 @@
+from script import Script
+
+
+class TestDispatcher(Script):
+    script_name = 'teuthology-dispatcher'
diff --git a/scripts/test/test_supervisor_.py b/scripts/test/test_supervisor_.py
new file mode 100644
index 0000000000..81298995c5
--- /dev/null
+++ b/scripts/test/test_supervisor_.py
@@ -0,0 +1,5 @@
+from script import Script
+
+
+class TestSupervisor(Script):
+    script_name = 'teuthology-supervisor'
diff --git a/scripts/test/test_worker.py b/scripts/test/test_worker.py
deleted file mode 100644
index 8e76c43a5c..0000000000
--- a/scripts/test/test_worker.py
+++ /dev/null
@@ -1,5 +0,0 @@
-from script import Script
-
-
-class TestWorker(Script):
-    script_name = 'teuthology-worker'
diff --git a/scripts/worker.py b/scripts/worker.py
deleted file mode 100644
index a3e12c20d7..0000000000
--- a/scripts/worker.py
+++ /dev/null
@@ -1,37 +0,0 @@
-import argparse
-
-import teuthology.worker
-
-
-def main():
-    teuthology.worker.main(parse_args())
-
-
-def parse_args():
-    parser = argparse.ArgumentParser(description="""
-Grab jobs from a beanstalk queue and run the teuthology tests they
-describe. One job is run at a time.
-""")
-    parser.add_argument(
-        '-v', '--verbose',
-        action='store_true', default=None,
-        help='be more verbose',
-    )
-    parser.add_argument(
-        '--archive-dir',
-        metavar='DIR',
-        help='path under which to archive results',
-        required=True,
-    )
-    parser.add_argument(
-        '-l', '--log-dir',
-        help='path in which to store logs',
-        required=True,
-    )
-    parser.add_argument(
-        '-t', '--tube',
-        help='which beanstalk tube to read jobs from',
-        required=True,
-    )
-
-    return parser.parse_args()
diff --git a/setup.cfg b/setup.cfg
index c6a5b4c892..73b0060302 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -84,6 +84,7 @@ console_scripts =
     teuthology-wait = scripts.wait:main
     teuthology-exporter = scripts.exporter:main
     teuthology-node-cleanup = scripts.node_cleanup:main
+    teuthology-supervisor = scripts.supervisor:main
 
 [options.extras_require]
 manhole =
diff --git a/teuthology/dispatcher/__init__.py b/teuthology/dispatcher/__init__.py
index fe309731d5..7cbaf7449b 100644
--- a/teuthology/dispatcher/__init__.py
+++ b/teuthology/dispatcher/__init__.py
@@ -17,11 +17,10 @@
     exporter,
     report,
     repo_utils,
-    worker,
 )
 from teuthology.config import config as teuth_config
 from teuthology.dispatcher import supervisor
-from teuthology.exceptions import SkipJob
+from teuthology.exceptions import BranchNotFoundError, CommitNotFoundError, SkipJob, MaxWhileTries
 from teuthology.lock import ops as lock_ops
 from teuthology import safepath
 
@@ -66,21 +65,10 @@ def load_config(archive_dir=None):
 
 
 def main(args):
-    # run dispatcher in job supervisor mode if --supervisor passed
-    if args["--supervisor"]:
-        return supervisor.main(args)
-
-    verbose = args["--verbose"]
-    tube = args["--tube"]
-    log_dir = args["--log-dir"]
-    archive_dir = args["--archive-dir"]
-    exit_on_empty_queue = args["--exit-on-empty-queue"]
-
-    if archive_dir is None:
-        archive_dir = teuth_config.archive_base
+    archive_dir = args.archive_dir or teuth_config.archive_base
 
     # Refuse to start more than one dispatcher per machine type
-    procs = find_dispatcher_processes().get(tube)
+    procs = find_dispatcher_processes().get(args.tube)
     if procs:
         raise RuntimeError(
             "There is already a teuthology-dispatcher process running:"
@@ -89,18 +77,18 @@ def main(args):
 
     # setup logging for disoatcher in {log_dir}
     loglevel = logging.INFO
-    if verbose:
+    if args.verbose:
         loglevel = logging.DEBUG
     logging.getLogger().setLevel(loglevel)
     log.setLevel(loglevel)
-    log_file_path = os.path.join(log_dir, f"dispatcher.{tube}.{os.getpid()}")
+    log_file_path = os.path.join(args.log_dir, f"dispatcher.{args.tube}.{os.getpid()}")
     setup_log_file(log_file_path)
     install_except_hook()
 
     load_config(archive_dir=archive_dir)
 
     connection = beanstalk.connect()
-    beanstalk.watch_tube(connection, tube)
+    beanstalk.watch_tube(connection, args.tube)
     result_proc = None
 
     if teuth_config.teuthology_path is None:
@@ -131,7 +119,7 @@ def main(args):
                 job_procs.remove(proc)
         job = connection.reserve(timeout=60)
         if job is None:
-            if exit_on_empty_queue and not job_procs:
+            if args.exit_on_empty_queue and not job_procs:
                 log.info("Queue is empty and no supervisor processes running; exiting!")
                 break
             continue
@@ -148,7 +136,7 @@ def main(args):
             keep_running = False
 
         try:
-            job_config, teuth_bin_path = worker.prep_job(
+            job_config, teuth_bin_path = prep_job(
                 job_config,
                 log_file_path,
                 archive_dir,
@@ -161,8 +149,7 @@ def main(args):
             job_config = lock_machines(job_config)
 
         run_args = [
-            os.path.join(teuth_bin_path, 'teuthology-dispatcher'),
-            '--supervisor',
+            os.path.join(teuth_bin_path, 'teuthology-supervisor'),
             '-v',
             '--bin-path', teuth_bin_path,
             '--archive-dir', archive_dir,
@@ -243,6 +230,82 @@ def match(proc):
     return procs
 
 
+def prep_job(job_config, log_file_path, archive_dir):
+    job_id = job_config['job_id']
+    safe_archive = safepath.munge(job_config['name'])
+    job_config['worker_log'] = log_file_path
+    archive_path_full = os.path.join(
+        archive_dir, safe_archive, str(job_id))
+    job_config['archive_path'] = archive_path_full
+
+    # If the teuthology branch was not specified, default to main and
+    # store that value.
+    teuthology_branch = job_config.get('teuthology_branch', 'main')
+    job_config['teuthology_branch'] = teuthology_branch
+    teuthology_sha1 = job_config.get('teuthology_sha1')
+    if not teuthology_sha1:
+        repo_url = repo_utils.build_git_url('teuthology', 'ceph')
+        try:
+            teuthology_sha1 = repo_utils.ls_remote(repo_url, teuthology_branch)
+        except Exception as exc:
+            log.exception(f"Could not get teuthology sha1 for branch {teuthology_branch}")
+            report.try_push_job_info(
+                job_config,
+                dict(status='dead', failure_reason=str(exc))
+            )
+            raise SkipJob()
+        if not teuthology_sha1:
+            reason = "Teuthology branch {} not found; marking job as dead".format(teuthology_branch)
+            log.error(reason)
+            report.try_push_job_info(
+                job_config,
+                dict(status='dead', failure_reason=reason)
+            )
+            raise SkipJob()
+        if teuth_config.teuthology_path is None:
+            log.info('Using teuthology sha1 %s', teuthology_sha1)
+
+    try:
+        if teuth_config.teuthology_path is not None:
+            teuth_path = teuth_config.teuthology_path
+        else:
+            teuth_path = repo_utils.fetch_teuthology(branch=teuthology_branch,
+                                          commit=teuthology_sha1)
+        # For the teuthology tasks, we look for suite_branch, and if we
+        # don't get that, we look for branch, and fall back to 'main'.
+        # last-in-suite jobs don't have suite_branch or branch set.
+        ceph_branch = job_config.get('branch', 'main')
+        suite_branch = job_config.get('suite_branch', ceph_branch)
+        suite_sha1 = job_config.get('suite_sha1')
+        suite_repo = job_config.get('suite_repo')
+        if suite_repo:
+            teuth_config.ceph_qa_suite_git_url = suite_repo
+        job_config['suite_path'] = os.path.normpath(os.path.join(
+            repo_utils.fetch_qa_suite(suite_branch, suite_sha1),
+            job_config.get('suite_relpath', ''),
+        ))
+    except (BranchNotFoundError, CommitNotFoundError) as exc:
+        log.exception("Requested version not found; marking job as dead")
+        report.try_push_job_info(
+            job_config,
+            dict(status='dead', failure_reason=str(exc))
+        )
+        raise SkipJob()
+    except MaxWhileTries as exc:
+        log.exception("Failed to fetch or bootstrap; marking job as dead")
+        report.try_push_job_info(
+            job_config,
+            dict(status='dead', failure_reason=str(exc))
+        )
+        raise SkipJob()
+
+    teuth_bin_path = os.path.join(teuth_path, 'virtualenv', 'bin')
+    if not os.path.isdir(teuth_bin_path):
+        raise RuntimeError("teuthology branch %s at %s not bootstrapped!" %
+                           (teuthology_branch, teuth_bin_path))
+    return job_config, teuth_bin_path
+
+
 def lock_machines(job_config):
     report.try_push_job_info(job_config, dict(status='running'))
     fake_ctx = supervisor.create_fake_context(job_config, block=True)
diff --git a/teuthology/dispatcher/supervisor.py b/teuthology/dispatcher/supervisor.py
index c003a4e620..2eb52f6637 100644
--- a/teuthology/dispatcher/supervisor.py
+++ b/teuthology/dispatcher/supervisor.py
@@ -24,17 +24,11 @@
 
 
 def main(args):
-
-    verbose = args["--verbose"]
-    archive_dir = args["--archive-dir"]
-    teuth_bin_path = args["--bin-path"]
-    config_file_path = args["--job-config"]
-
-    with open(config_file_path, 'r') as config_file:
+    with open(args.job_config, 'r') as config_file:
         job_config = yaml.safe_load(config_file)
 
     loglevel = logging.INFO
-    if verbose:
+    if args.verbose:
         loglevel = logging.DEBUG
     logging.getLogger().setLevel(loglevel)
     log.setLevel(loglevel)
@@ -57,7 +51,7 @@ def main(args):
                 reimage(job_config)
         else:
             reimage(job_config)
-        with open(config_file_path, 'w') as f:
+        with open(args.job_config, 'w') as f:
             yaml.safe_dump(job_config, f, default_flow_style=False)
 
     try:
@@ -66,16 +60,16 @@ def main(args):
             with exporter.JobTime.labels(suite).time():
                 return run_job(
                     job_config,
-                    teuth_bin_path,
-                    archive_dir,
-                    verbose
+                    args.bin_path,
+                    args.archive_dir,
+                    args.verbose
                 )
         else:
             return run_job(
                 job_config,
-                teuth_bin_path,
-                archive_dir,
-                verbose
+                args.bin_path,
+                args.archive_dir,
+                args.verbose
             )
     except SkipJob:
         return 0
diff --git a/teuthology/dispatcher/test/test_dispatcher.py b/teuthology/dispatcher/test/test_dispatcher.py
new file mode 100644
index 0000000000..e7c59d8bd2
--- /dev/null
+++ b/teuthology/dispatcher/test/test_dispatcher.py
@@ -0,0 +1,174 @@
+import datetime
+import os
+import pytest
+
+from unittest.mock import patch, Mock, MagicMock
+
+from teuthology import dispatcher
+from teuthology.config import FakeNamespace
+from teuthology.contextutil import MaxWhileTries
+
+
+class TestDispatcher(object):
+    @pytest.fixture(autouse=True)
+    def setup_method(self, tmp_path):
+        self.ctx = FakeNamespace()
+        self.ctx.verbose = True
+        self.ctx.archive_dir = str(tmp_path / "archive/dir")
+        self.ctx.log_dir = str(tmp_path / "log/dir")
+        self.ctx.tube = 'tube'
+
+    @patch("os.path.exists")
+    def test_restart_file_path_doesnt_exist(self, m_exists):
+        m_exists.return_value = False
+        result = dispatcher.sentinel(dispatcher.restart_file_path)
+        assert not result
+
+    @patch("os.path.getmtime")
+    @patch("os.path.exists")
+    def test_needs_restart(self, m_exists, m_getmtime):
+        m_exists.return_value = True
+        now = datetime.datetime.now(datetime.timezone.utc)
+        m_getmtime.return_value = (now + datetime.timedelta(days=1)).timestamp()
+        assert dispatcher.sentinel(dispatcher.restart_file_path)
+
+    @patch("os.path.getmtime")
+    @patch("os.path.exists")
+    def test_does_not_need_restart(self, m_exists, m_getmtime):
+        m_exists.return_value = True
+        now = datetime.datetime.now(datetime.timezone.utc)
+        m_getmtime.return_value = (now - datetime.timedelta(days=1)).timestamp()
+        assert not dispatcher.sentinel(dispatcher.restart_file_path)
+
+    @patch("teuthology.repo_utils.ls_remote")
+    @patch("os.path.isdir")
+    @patch("teuthology.repo_utils.fetch_teuthology")
+    @patch("teuthology.dispatcher.teuth_config")
+    @patch("teuthology.repo_utils.fetch_qa_suite")
+    def test_prep_job(self, m_fetch_qa_suite, m_teuth_config,
+            m_fetch_teuthology, m_isdir, m_ls_remote):
+        config = dict(
+            name="the_name",
+            job_id="1",
+            suite_sha1="suite_hash",
+        )
+        m_fetch_teuthology.return_value = '/teuth/path'
+        m_fetch_qa_suite.return_value = '/suite/path'
+        m_ls_remote.return_value = 'teuth_hash'
+        m_isdir.return_value = True
+        m_teuth_config.teuthology_path = None
+        got_config, teuth_bin_path = dispatcher.prep_job(
+            config,
+            self.ctx.log_dir,
+            self.ctx.archive_dir,
+        )
+        assert got_config['worker_log'] == self.ctx.log_dir
+        assert got_config['archive_path'] == os.path.join(
+            self.ctx.archive_dir,
+            config['name'],
+            config['job_id'],
+        )
+        assert got_config['teuthology_branch'] == 'main'
+        m_fetch_teuthology.assert_called_once_with(branch='main', commit='teuth_hash')
+        assert teuth_bin_path == '/teuth/path/virtualenv/bin'
+        m_fetch_qa_suite.assert_called_once_with('main', 'suite_hash')
+        assert got_config['suite_path'] == '/suite/path'
+
+    def build_fake_jobs(self, m_connection, m_job, job_bodies):
+        """
+        Given patched copies of:
+            beanstalkc.Connection
+            beanstalkc.Job
+        And a list of basic job bodies, return a list of mocked Job objects
+        """
+        # Make sure instantiating m_job returns a new object each time
+        jobs = []
+        job_id = 0
+        for job_body in job_bodies:
+            job_id += 1
+            job = MagicMock(conn=m_connection, jid=job_id, body=job_body)
+            job.jid = job_id
+            job.body = job_body
+            jobs.append(job)
+        return jobs
+
+    @patch("teuthology.dispatcher.find_dispatcher_processes")
+    @patch("teuthology.repo_utils.ls_remote")
+    @patch("teuthology.dispatcher.report.try_push_job_info")
+    @patch("teuthology.dispatcher.supervisor.run_job")
+    @patch("beanstalkc.Job", autospec=True)
+    @patch("teuthology.repo_utils.fetch_qa_suite")
+    @patch("teuthology.repo_utils.fetch_teuthology")
+    @patch("teuthology.dispatcher.beanstalk.watch_tube")
+    @patch("teuthology.dispatcher.beanstalk.connect")
+    @patch("os.path.isdir", return_value=True)
+    @patch("teuthology.dispatcher.setup_log_file")
+    def test_main_loop(
+        self, m_setup_log_file, m_isdir, m_connect, m_watch_tube,
+        m_fetch_teuthology, m_fetch_qa_suite, m_job, m_run_job,
+        m_try_push_job_info, m_ls_remote, m_find_dispatcher_processes,
+                       ):
+        m_find_dispatcher_processes.return_value = {}
+        m_connection = Mock()
+        jobs = self.build_fake_jobs(
+            m_connection,
+            m_job,
+            [
+                'name: name\nfoo: bar',
+                'name: name\nstop_worker: true',
+            ],
+        )
+        m_connection.reserve.side_effect = jobs
+        m_connect.return_value = m_connection
+        dispatcher.main(self.ctx)
+        # There should be one reserve call per item in the jobs list
+        expected_reserve_calls = [
+            dict(timeout=60) for i in range(len(jobs))
+        ]
+        got_reserve_calls = [
+            call[1] for call in m_connection.reserve.call_args_list
+        ]
+        assert got_reserve_calls == expected_reserve_calls
+        for job in jobs:
+            job.bury.assert_called_once_with()
+            job.delete.assert_called_once_with()
+
+    @patch("teuthology.dispatcher.find_dispatcher_processes")
+    @patch("teuthology.repo_utils.ls_remote")
+    @patch("teuthology.dispatcher.report.try_push_job_info")
+    @patch("teuthology.dispatcher.supervisor.run_job")
+    @patch("beanstalkc.Job", autospec=True)
+    @patch("teuthology.repo_utils.fetch_qa_suite")
+    @patch("teuthology.repo_utils.fetch_teuthology")
+    @patch("teuthology.dispatcher.beanstalk.watch_tube")
+    @patch("teuthology.dispatcher.beanstalk.connect")
+    @patch("os.path.isdir", return_value=True)
+    @patch("teuthology.dispatcher.setup_log_file")
+    def test_main_loop_13925(
+        self, m_setup_log_file, m_isdir, m_connect, m_watch_tube,
+        m_fetch_teuthology, m_fetch_qa_suite, m_job, m_run_job,
+        m_try_push_job_info, m_ls_remote, m_find_dispatcher_processes,
+                       ):
+        m_find_dispatcher_processes.return_value = {}
+        m_connection = Mock()
+        jobs = self.build_fake_jobs(
+            m_connection,
+            m_job,
+            [
+                'name: name',
+                'name: name\nstop_worker: true',
+            ],
+        )
+        m_connection.reserve.side_effect = jobs
+        m_connect.return_value = m_connection
+        m_fetch_qa_suite.side_effect = [
+            '/suite/path',
+            MaxWhileTries(),
+            MaxWhileTries(),
+        ]
+        dispatcher.main(self.ctx)
+        assert len(m_run_job.call_args_list) == 0
+        assert len(m_try_push_job_info.call_args_list) == len(jobs)
+        for i in range(len(jobs)):
+            push_call = m_try_push_job_info.call_args_list[i]
+            assert push_call[0][1]['status'] == 'dead'
diff --git a/teuthology/dispatcher/test/test_supervisor.py b/teuthology/dispatcher/test/test_supervisor.py
new file mode 100644
index 0000000000..2b422c07b1
--- /dev/null
+++ b/teuthology/dispatcher/test/test_supervisor.py
@@ -0,0 +1,117 @@
+from subprocess import DEVNULL
+from unittest.mock import patch, Mock, MagicMock
+
+from teuthology.dispatcher import supervisor
+
+
+class TestSuperviser(object):
+    @patch("teuthology.dispatcher.supervisor.run_with_watchdog")
+    @patch("teuthology.dispatcher.supervisor.teuth_config")
+    @patch("subprocess.Popen")
+    @patch("os.environ")
+    @patch("os.mkdir")
+    @patch("yaml.safe_dump")
+    @patch("tempfile.NamedTemporaryFile")
+    def test_run_job_with_watchdog(self, m_tempfile, m_safe_dump, m_mkdir,
+                                   m_environ, m_popen, m_t_config,
+                                   m_run_watchdog):
+        config = {
+            "suite_path": "suite/path",
+            "config": {"foo": "bar"},
+            "verbose": True,
+            "owner": "the_owner",
+            "archive_path": "archive/path",
+            "name": "the_name",
+            "description": "the_description",
+            "job_id": "1",
+        }
+        m_tmp = MagicMock()
+        temp_file = Mock()
+        temp_file.name = "the_name"
+        m_tmp.__enter__.return_value = temp_file
+        m_tempfile.return_value = m_tmp
+        m_p = Mock()
+        m_p.returncode = 0
+        m_popen.return_value = m_p
+        m_t_config.results_server = True
+        supervisor.run_job(config, "teuth/bin/path", "archive/dir", verbose=False)
+        m_run_watchdog.assert_called_with(m_p, config)
+        expected_args = [
+            'teuth/bin/path/teuthology',
+            '-v',
+            '--owner', 'the_owner',
+            '--archive', 'archive/path',
+            '--name', 'the_name',
+            '--description',
+            'the_description',
+            '--',
+            "archive/path/orig.config.yaml",
+        ]
+        m_popen.assert_called_with(args=expected_args, stderr=DEVNULL, stdout=DEVNULL)
+
+    @patch("time.sleep")
+    @patch("teuthology.dispatcher.supervisor.teuth_config")
+    @patch("subprocess.Popen")
+    @patch("os.environ")
+    @patch("os.mkdir")
+    @patch("yaml.safe_dump")
+    @patch("tempfile.NamedTemporaryFile")
+    def test_run_job_no_watchdog(self, m_tempfile, m_safe_dump, m_mkdir,
+                                 m_environ, m_popen, m_t_config,
+                                 m_sleep):
+        config = {
+            "suite_path": "suite/path",
+            "config": {"foo": "bar"},
+            "verbose": True,
+            "owner": "the_owner",
+            "archive_path": "archive/path",
+            "name": "the_name",
+            "description": "the_description",
+            "job_id": "1",
+        }
+        m_tmp = MagicMock()
+        temp_file = Mock()
+        temp_file.name = "the_name"
+        m_tmp.__enter__.return_value = temp_file
+        m_tempfile.return_value = m_tmp
+        env = dict(PYTHONPATH="python/path")
+        m_environ.copy.return_value = env
+        m_p = Mock()
+        m_p.returncode = 1
+        m_popen.return_value = m_p
+        m_t_config.results_server = False
+        supervisor.run_job(config, "teuth/bin/path", "archive/dir", verbose=False)
+
+    @patch("teuthology.dispatcher.supervisor.report.try_push_job_info")
+    @patch("time.sleep")
+    def test_run_with_watchdog_no_reporting(self, m_sleep, m_try_push):
+        config = {
+            "name": "the_name",
+            "job_id": "1",
+            "archive_path": "archive/path",
+            "teuthology_branch": "main"
+        }
+        process = Mock()
+        process.poll.return_value = "not None"
+        supervisor.run_with_watchdog(process, config)
+        m_try_push.assert_called_with(
+            dict(name=config["name"], job_id=config["job_id"]),
+            dict(status='dead')
+        )
+
+    @patch("subprocess.Popen")
+    @patch("time.sleep")
+    @patch("teuthology.dispatcher.supervisor.report.try_push_job_info")
+    def test_run_with_watchdog_with_reporting(self, m_tpji, m_sleep, m_popen):
+        config = {
+            "name": "the_name",
+            "job_id": "1",
+            "archive_path": "archive/path",
+            "teuthology_branch": "jewel"
+        }
+        process = Mock()
+        process.poll.return_value = "not None"
+        m_proc = Mock()
+        m_proc.poll.return_value = "not None"
+        m_popen.return_value = m_proc
+        supervisor.run_with_watchdog(process, config)