Repository: PrefectHQ/prefect

Issue:
LocalDaskExecutor with threaded scheduler fails to run mapped tasks with timeout on linux
## Description
The following example: https://github.com/mianos/pbug demonstrates prefect/dask not scheduling a trivial task.
python slow.py --range=10 --sleep_time=3 > ubuntu2004run.txt

https://github.com/mianos/pbug/blob/main/ubuntu/ubuntu2004run.txt
This is on a 8 core local physical machine with no load.


## Expected Behavior
The messages '==== IN TASK XX Sleeping 3' should appear for each item in the range.
Note that the task runner is scheduling the tasks from the lines 'Task 'slow_task[3]': Starting task run...'
The result set should include results for all items.
The task runner hits the timeout (30 seconds) and everything gives up


## Reproduction
https://github.com/mianos/pbug/blob/main/slow.py
checkout, make a venv, pip install -r requirements.txt, run as above.

Sometimes this works but it seems the faster the machine the more it fails.

## Environment
I have tested it on redhat, CentOS Linux release 7.7.1908 (Core)

The results provided above are from Ununtu 20.04
Run and all versions are provided here:
https://github.com/mianos/pbug/tree/main/ubuntu

Fails on python 3.85 and python 3.6.8

{
  "config_overrides": {},
  "env_vars": [],
  "system_information": {
    "platform": "Linux-5.4.0-48-generic-x86_64-with-glibc2.29",
    "prefect_version": "0.13.3",
    "python_version": "3.8.5"
  }
}

I can provide SSH access to some machines demonstrating the failure if needed.

Discussion:

Patch:
diff --git a/changes/pr3256.yaml b/changes/pr3256.yaml
new file mode 100644
index 0000000000..039370f0ee
--- /dev/null
+++ b/changes/pr3256.yaml
@@ -0,0 +1,2 @@
+fix:
+  - "Fix multiprocess based timeout handler on linux ` - [#3526](https://github.com/PrefectHQ/prefect/pull/3526)"
diff --git a/changes/pr3527.yaml b/changes/pr3527.yaml
new file mode 100644
index 0000000000..4c30970520
--- /dev/null
+++ b/changes/pr3527.yaml
@@ -0,0 +1,5 @@
+enhancement:
+  - "Add `LocalRun` implementation for `run_config` based flows - [#3527](https://github.com/PrefectHQ/prefect/pull/3527)"
+
+fix:
+  - "Add labels from `flow.storage` for `run_config` based flows - [#3527](https://github.com/PrefectHQ/prefect/pull/3527)"
diff --git a/changes/pr3537.yaml b/changes/pr3537.yaml
new file mode 100644
index 0000000000..96a9dcfc4e
--- /dev/null
+++ b/changes/pr3537.yaml
@@ -0,0 +1,2 @@
+enhancement:
+  - "Add `DockerRun` implementation for `run_config` based flows - [#3537](https://github.com/PrefectHQ/prefect/pull/3537)"
diff --git a/docs/core/advanced_tutorials/slack-notifications.md b/docs/core/advanced_tutorials/slack-notifications.md
index e31c138270..e28f3beb19 100644
--- a/docs/core/advanced_tutorials/slack-notifications.md
+++ b/docs/core/advanced_tutorials/slack-notifications.md
@@ -51,7 +51,7 @@ After making your decisions, click the green "Authorize" button to proceed. Assu
 
 ![slack hook authorized](/slack_page2.png){.viz-xl .viz-padded}
 
-That's it! Save the URL in your secure location of choice; soon, you will be able to automatically import this into your database of Prefect Secrets.
+That's it! Save the URL in your secure location of choice.
 
 ::: tip Multiple Installations
 It's perfectly OK to integrate the Prefect App multiple times into the same workspace; for example, you and a coworker who both use Prefect might want customized notifications. Follow these steps as many times as you desire, but make sure to keep track of which channel each URL is attached to!
@@ -59,7 +59,9 @@ It's perfectly OK to integrate the Prefect App multiple times into the same work
 
 ## Using your URL to get notifications
 
-Sooner rather than later, you will be able to store your slack webhook URL in your secure database of Prefect Secrets under `"SLACK_WEBHOOK_URL"`; until then, you need to include it in the `[context.secrets]` section of your prefect configuration file. To do so, create a file `~/.prefect/config.toml` and place the following into it:
+You can store your slack webhook URL in your secure database of Prefect Secrets under `"SLACK_WEBHOOK_URL"`. 
+
+Alternatively, for testing and development, you can include it in the `[context.secrets]` section of your prefect configuration file. To do so, create a file `~/.prefect/config.toml` and place the following into it:
 
 ```
 [context.secrets]
@@ -67,7 +69,7 @@ SLACK_WEBHOOK_URL = "https://hooks.slack.com/services/XXXXXXXXX/xxxxxxxxx/XXXXXX
 ```
 
 ::: warning Storing Secrets
-This method of storing secrets is intended _only_ for local development and testing!
+This method of storing secrets is intended _only_ for local development and testing! For production usage, use a Prefect Secret instead.
 :::
 
 Almost there - all that's left is to actually hook up the `slack_notifier` state handler to your favorite task or flow! All task and flow initializations accept an optional `state_handler` keyword argument. This should consist of a _list_ of "state handler" callables with call signature
diff --git a/docs/outline.toml b/docs/outline.toml
index 1c7e96e5b7..f695cc8e2f 100644
--- a/docs/outline.toml
+++ b/docs/outline.toml
@@ -235,7 +235,7 @@ classes = ["DaskKubernetesEnvironment", "DaskCloudProviderEnvironment", "Fargate
 [pages.run_configs]
 title = "Run Configuration"
 module = "prefect.run_configs"
-classes = ["RunConfig", "KubernetesRun"]
+classes = ["RunConfig", "LocalRun", "DockerRun", "KubernetesRun"]
 
 [pages.tasks.control_flow]
 title = "Control Flow Tasks"
@@ -545,7 +545,7 @@ functions = ["diagnostic_info", "flow_information", "system_information", "confi
 [pages.utilities.executors]
 title = "Executors"
 module = "prefect.utilities.executors"
-functions = ["timeout_handler"]
+functions = ["run_task_with_timeout"]
 
 [pages.utilities.gcp]
 title = "Google Utilities"
diff --git a/src/prefect/agent/docker/agent.py b/src/prefect/agent/docker/agent.py
index 86fb1b6f36..de74321dcd 100644
--- a/src/prefect/agent/docker/agent.py
+++ b/src/prefect/agent/docker/agent.py
@@ -8,6 +8,8 @@ from typing import TYPE_CHECKING, Dict, Iterable, List, Tuple
 
 from prefect import config, context
 from prefect.agent import Agent
+from prefect.run_configs import DockerRun
+from prefect.serialization.run_config import RunConfigSchema
 from prefect.utilities.agent import get_flow_image, get_flow_run_command
 from prefect.utilities.docker_util import get_docker_ip
 from prefect.utilities.graphql import GraphQLResult
@@ -349,8 +351,22 @@ class DockerAgent(Agent):
         # the 'import prefect' time low
         import docker
 
+        if getattr(flow_run.flow, "run_config", None) is not None:
+            run_config = RunConfigSchema().load(flow_run.flow.run_config)
+            if not isinstance(run_config, DockerRun):
+                self.logger.error(
+                    "Flow run %s has a `run_config` of type `%s`, only `DockerRun` is supported",
+                    flow_run.id,
+                    type(run_config).__name__,
+                )
+                raise TypeError(
+                    "Unsupported RunConfig type: %s" % type(run_config).__name__
+                )
+        else:
+            run_config = None
+
         image = get_flow_image(flow_run=flow_run)
-        env_vars = self.populate_env_vars(flow_run=flow_run)
+        env_vars = self.populate_env_vars(flow_run, run_config=run_config)
 
         if not self.no_pull and len(image.split("/")) > 1:
             self.logger.info("Pulling image {}...".format(image))
@@ -444,12 +460,15 @@ class DockerAgent(Agent):
         proc.start()
         self.processes.append(proc)
 
-    def populate_env_vars(self, flow_run: GraphQLResult) -> dict:
+    def populate_env_vars(
+        self, flow_run: GraphQLResult, run_config: DockerRun = None
+    ) -> dict:
         """
         Populate metadata and variables in the environment variables for a flow run
 
         Args:
             - flow_run (GraphQLResult): A flow run object
+            - run_config (DockerRun, optional): The `run_config` for the flow, if any.
 
         Returns:
             - dict: a dictionary representing the populated environment variables
@@ -459,19 +478,39 @@ class DockerAgent(Agent):
         else:
             api = config.cloud.api
 
-        return {
-            "PREFECT__CLOUD__API": api,
-            "PREFECT__CLOUD__AUTH_TOKEN": config.cloud.agent.auth_token,
-            "PREFECT__CLOUD__AGENT__LABELS": str(self.labels),
-            "PREFECT__CONTEXT__FLOW_RUN_ID": flow_run.id,  # type: ignore
-            "PREFECT__CONTEXT__FLOW_ID": flow_run.flow.id,  # type: ignore
-            "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
-            "PREFECT__LOGGING__LOG_TO_CLOUD": str(self.log_to_cloud).lower(),
-            "PREFECT__LOGGING__LEVEL": config.logging.level,
-            "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
-            "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
-            **self.env_vars,
-        }
+        env = {}
+        # Populate environment variables, later sources overriding
+
+        # 1. Logging config (optional)
+        # We only set this on non-DockerRun runs for backwards compatibility.
+        # In the future we don't want the agent to set logging config on a flow
+        # run unless explicitly asked.  We do this early on so later config
+        # sources can override
+        if run_config is None:
+            env.update({"PREFECT__LOGGING__LEVEL": config.logging.level})
+
+        # 2. Values set on the agent via `--env`
+        env.update(self.env_vars)
+
+        # 3. Values set on a DockerRun RunConfig (if present
+        if run_config is not None and run_config.env is not None:
+            env.update(run_config.env)
+
+        # 4. Non-overrideable required env vars
+        env.update(
+            {
+                "PREFECT__CLOUD__API": api,
+                "PREFECT__CLOUD__AUTH_TOKEN": config.cloud.agent.auth_token,
+                "PREFECT__CLOUD__AGENT__LABELS": str(self.labels),
+                "PREFECT__CONTEXT__FLOW_RUN_ID": flow_run.id,  # type: ignore
+                "PREFECT__CONTEXT__FLOW_ID": flow_run.flow.id,  # type: ignore
+                "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
+                "PREFECT__LOGGING__LOG_TO_CLOUD": str(self.log_to_cloud).lower(),
+                "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
+                "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
+            }
+        )
+        return env
 
 
 if __name__ == "__main__":
diff --git a/src/prefect/agent/local/agent.py b/src/prefect/agent/local/agent.py
index 222fa55ce7..61cb6ba95a 100644
--- a/src/prefect/agent/local/agent.py
+++ b/src/prefect/agent/local/agent.py
@@ -6,8 +6,10 @@ from typing import Iterable, List
 
 from prefect import config
 from prefect.agent import Agent
-from prefect.environments.storage import GCS, S3, Azure, Local, GitHub, GitLab, Webhook
+from prefect.environments.storage import Docker
+from prefect.run_configs import LocalRun
 from prefect.serialization.storage import StorageSchema
+from prefect.serialization.run_config import RunConfigSchema
 from prefect.utilities.agent import get_flow_run_command
 from prefect.utilities.graphql import GraphQLResult
 
@@ -136,29 +138,37 @@ class LocalAgent(Agent):
         """
         self.logger.info("Deploying flow run {}".format(flow_run.id))  # type: ignore
 
-        if not isinstance(
-            StorageSchema().load(flow_run.flow.storage),
-            (Local, Azure, GCS, S3, GitHub, GitLab, Webhook),
-        ):
+        storage = StorageSchema().load(flow_run.flow.storage)
+        if isinstance(storage, Docker):
             self.logger.error(
-                "Storage for flow run {} is not a supported type.".format(flow_run.id)
+                "Flow run %s has an unsupported storage type: `%s`",
+                flow_run.id,
+                type(storage).__name__,
             )
-            raise ValueError("Unsupported Storage type")
-
-        env_vars = self.populate_env_vars(flow_run=flow_run)
-        current_env = os.environ.copy()
-        current_env.update(env_vars)
-
-        python_path = []
-        if current_env.get("PYTHONPATH"):
-            python_path.append(current_env.get("PYTHONPATH"))
-
-        python_path.append(os.getcwd())
-
-        if self.import_paths:
-            python_path += self.import_paths
-
-        current_env["PYTHONPATH"] = ":".join(python_path)
+            raise TypeError("Unsupported Storage type: %s" % type(storage).__name__)
+
+        # If the flow is using a run_config, load it
+        if getattr(flow_run.flow, "run_config", None) is not None:
+            run_config = RunConfigSchema().load(flow_run.flow.run_config)
+            if not isinstance(run_config, LocalRun):
+                self.logger.error(
+                    "Flow run %s has a `run_config` of type `%s`, only `LocalRun` is supported",
+                    flow_run.id,
+                    type(run_config).__name__,
+                )
+                raise TypeError(
+                    "Unsupported RunConfig type: %s" % type(run_config).__name__
+                )
+        else:
+            run_config = None
+
+        env = self.populate_env_vars(flow_run, run_config=run_config)
+
+        working_dir = None if run_config is None else run_config.working_dir
+        if working_dir and not os.path.exists(working_dir):
+            msg = f"Flow run {flow_run.id} has a nonexistent `working_dir` configured: {working_dir}"
+            self.logger.error(msg)
+            raise ValueError(msg)
 
         stdout = sys.stdout if self.show_flow_logs else DEVNULL
 
@@ -171,7 +181,8 @@ class LocalAgent(Agent):
             get_flow_run_command(flow_run).split(" "),
             stdout=stdout,
             stderr=STDOUT,
-            env=current_env,
+            env=env,
+            cwd=working_dir,
         )
 
         self.processes.add(p)
@@ -181,30 +192,65 @@ class LocalAgent(Agent):
 
         return "PID: {}".format(p.pid)
 
-    def populate_env_vars(self, flow_run: GraphQLResult) -> dict:
+    def populate_env_vars(
+        self, flow_run: GraphQLResult, run_config: LocalRun = None
+    ) -> dict:
         """
         Populate metadata and variables in the environment variables for a flow run
 
         Args:
             - flow_run (GraphQLResult): A flow run object
+            - run_config (LocalRun, optional): The `run_config` for the flow, if any.
 
         Returns:
             - dict: a dictionary representing the populated environment variables
         """
-        all_vars = {
-            "PREFECT__CLOUD__API": config.cloud.api,
-            "PREFECT__CLOUD__AUTH_TOKEN": self.client._api_token,
-            "PREFECT__CLOUD__AGENT__LABELS": str(self.labels),
-            "PREFECT__CONTEXT__FLOW_RUN_ID": flow_run.id,  # type: ignore
-            "PREFECT__CONTEXT__FLOW_ID": flow_run.flow.id,  # type: ignore
-            "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
-            "PREFECT__LOGGING__LOG_TO_CLOUD": str(self.log_to_cloud).lower(),
-            "PREFECT__LOGGING__LEVEL": config.logging.level,
-            "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
-            "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
-            **self.env_vars,
-        }
-        return {k: v for k, v in all_vars.items() if v is not None}
+        # Environment variables come from (later options override):
+
+        # 1. Local environment
+        env = os.environ.copy()
+
+        # 2. Logging config
+        env.update({"PREFECT__LOGGING__LEVEL": config.logging.level})
+
+        # 3. PYTHONPATH
+        # XXX: I'm 99% sure this isn't needed, a new Python process should
+        # always add the CWD as the first entry in sys.path
+        python_path = [
+            run_config.working_dir
+            if run_config is not None and run_config.working_dir
+            else os.getcwd()
+        ]
+        if os.environ.get("PYTHONPATH"):
+            python_path.append(os.environ["PYTHONPATH"])
+        if self.import_paths:
+            python_path.extend(self.import_paths)
+        env["PYTHONPATH"] = ":".join(python_path)
+
+        # 4. Values set on the agent via `--env`
+        env.update(self.env_vars)
+
+        # 5. Values set on a LocalRun RunConfig (if present
+        if run_config is not None and run_config.env is not None:
+            env.update(run_config.env)
+
+        # 6. Non-overrideable required env vars
+        env.update(
+            {
+                "PREFECT__CLOUD__API": config.cloud.api,
+                "PREFECT__CLOUD__AUTH_TOKEN": self.client._api_token,
+                "PREFECT__CLOUD__AGENT__LABELS": str(self.labels),
+                "PREFECT__CONTEXT__FLOW_RUN_ID": flow_run.id,  # type: ignore
+                "PREFECT__CONTEXT__FLOW_ID": flow_run.flow.id,  # type: ignore
+                "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
+                "PREFECT__LOGGING__LOG_TO_CLOUD": str(self.log_to_cloud).lower(),
+                "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
+                "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
+            }
+        )
+
+        # Filter out None values
+        return {k: v for k, v in env.items() if v is not None}
 
     @staticmethod
     def generate_supervisor_conf(
diff --git a/src/prefect/core/flow.py b/src/prefect/core/flow.py
index 8adcb9151d..1eeb54c2d0 100644
--- a/src/prefect/core/flow.py
+++ b/src/prefect/core/flow.py
@@ -1539,7 +1539,10 @@ class Flow:
             "logging.log_to_cloud": log_to_cloud,
         }
         with set_temporary_config(temp_config):
-            labels = list(self.environment.labels) if self.environment.labels else []
+            if self.run_config is not None:
+                labels = list(self.run_config.labels or ())
+            else:
+                labels = list(self.environment.labels or ())
             agent = prefect.agent.local.LocalAgent(
                 labels=labels, show_flow_logs=show_flow_logs
             )
@@ -1606,10 +1609,10 @@ class Flow:
             self.storage = get_default_storage_class()(**kwargs)
 
         # add auto-labels for various types of storage
-        self.environment.labels.update(self.storage.labels)
-
-        if labels:
-            self.environment.labels.update(labels)
+        for obj in [self.environment, self.run_config]:
+            if obj is not None:
+                obj.labels.update(self.storage.labels)
+                obj.labels.update(labels or ())
 
         # register the flow with a default result handler if one not provided
         if not self.result:
diff --git a/src/prefect/engine/task_runner.py b/src/prefect/engine/task_runner.py
index b0951f38ea..f009cae856 100644
--- a/src/prefect/engine/task_runner.py
+++ b/src/prefect/engine/task_runner.py
@@ -1,5 +1,6 @@
 from contextlib import redirect_stdout
 from dask.base import tokenize
+from contextlib import AbstractContextManager
 from typing import (
     Any,
     Callable,
@@ -40,6 +41,7 @@ from prefect.utilities.executors import (
     run_with_heartbeat,
     tail_recursive,
 )
+from prefect.utilities.compatibility import nullcontext
 
 
 TaskRunnerInitializeResult = NamedTuple(
@@ -309,7 +311,7 @@ class TaskRunner(Runner):
             raise exc
 
         except Exception as exc:
-            msg = "Task '{name}': unexpected error while running task: {exc}".format(
+            msg = "Task '{name}': Unexpected error while running task: {exc}".format(
                 name=context["task_full_name"], exc=repr(exc)
             )
             self.logger.exception(msg)
@@ -325,7 +327,8 @@ class TaskRunner(Runner):
             # that any run-context, including task-run-ids, are respected
             with prefect.context(context):
                 self.logger.info(
-                    "Task '{name}': finished task run for task with final state: '{state}'".format(
+                    "Task '{name}': Finished task run for task with final state: "
+                    "'{state}'".format(
                         name=context["task_full_name"], state=type(state).__name__
                     )
                 )
@@ -361,7 +364,8 @@ class TaskRunner(Runner):
 
         if not all(s.is_finished() for s in all_states):
             self.logger.debug(
-                "Task '{name}': not all upstream states are finished; ending run.".format(
+                "Task '{name}': Not all upstream states are finished; "
+                "ending run.".format(
                     name=prefect.context.get("task_full_name", self.task.name)
                 )
             )
@@ -504,7 +508,8 @@ class TaskRunner(Runner):
         # Exceptions are trapped and turned into TriggerFailed states
         except Exception as exc:
             self.logger.exception(
-                "Task '{name}': unexpected error while evaluating task trigger: {exc}".format(
+                "Task '{name}': Unexpected error while evaluating task trigger: "
+                "{exc}".format(
                     exc=repr(exc),
                     name=prefect.context.get("task_full_name", self.task.name),
                 )
@@ -564,7 +569,7 @@ class TaskRunner(Runner):
         # this task is already finished
         elif state.is_finished():
             self.logger.debug(
-                "Task '{name}': task is already finished.".format(
+                "Task '{name}': Task is already finished.".format(
                     name=prefect.context.get("task_full_name", self.task.name)
                 )
             )
@@ -573,7 +578,8 @@ class TaskRunner(Runner):
         # this task is not pending
         else:
             self.logger.debug(
-                "Task '{name}' is not ready to run or state was unrecognized ({state}).".format(
+                "Task '{name}': Task is not ready to run or state was unrecognized "
+                "({state}).".format(
                     name=prefect.context.get("task_full_name", self.task.name),
                     state=state,
                 )
@@ -601,7 +607,8 @@ class TaskRunner(Runner):
             # handle case where no start_time is set
             if state.start_time is None:
                 self.logger.debug(
-                    "Task '{name}' is scheduled without a known start_time; ending run.".format(
+                    "Task '{name}' is scheduled without a known start_time; "
+                    "ending run.".format(
                         name=prefect.context.get("task_full_name", self.task.name)
                     )
                 )
@@ -610,7 +617,8 @@ class TaskRunner(Runner):
             # handle case where start time is in the future
             elif state.start_time and state.start_time > pendulum.now("utc"):
                 self.logger.debug(
-                    "Task '{name}': start_time has not been reached; ending run.".format(
+                    "Task '{name}': start_time has not been reached; "
+                    "ending run.".format(
                         name=prefect.context.get("task_full_name", self.task.name)
                     )
                 )
@@ -622,8 +630,8 @@ class TaskRunner(Runner):
         self, state: State, upstream_states: Dict[Edge, State]
     ) -> Dict[str, Result]:
         """
-        Given the task's current state and upstream states, generates the inputs for this task.
-        Upstream state result values are used.
+        Given the task's current state and upstream states, generates the inputs for
+        this task. Upstream state result values are used.
 
         Args:
             - state (State): the task's current state.
@@ -753,7 +761,7 @@ class TaskRunner(Runner):
 
         if self.task.cache_for is not None:
             self.logger.warning(
-                "Task '{name}': can't use cache because it "
+                "Task '{name}': Can't use cache because it "
                 "is now invalid".format(
                     name=prefect.context.get("task_full_name", self.task.name)
                 )
@@ -778,7 +786,7 @@ class TaskRunner(Runner):
         """
         if not state.is_pending():
             self.logger.debug(
-                "Task '{name}': can't set state to Running because it "
+                "Task '{name}': Can't set state to Running because it "
                 "isn't Pending; ending run.".format(
                     name=prefect.context.get("task_full_name", self.task.name)
                 )
@@ -809,7 +817,7 @@ class TaskRunner(Runner):
         """
         if not state.is_running():
             self.logger.debug(
-                "Task '{name}': can't run task because it's not in a "
+                "Task '{name}': Can't run task because it's not in a "
                 "Running state; ending run.".format(
                     name=prefect.context.get("task_full_name", self.task.name)
                 )
@@ -825,17 +833,20 @@ class TaskRunner(Runner):
                     name=prefect.context.get("task_full_name", self.task.name)
                 )
             )
-            timeout_handler = prefect.utilities.executors.timeout_handler
-            if getattr(self.task, "log_stdout", False):
-                with redirect_stdout(  # type: ignore
-                    prefect.utilities.logging.RedirectToLog(self.logger)
-                ):
-                    value = timeout_handler(
-                        self.task.run, timeout=self.task.timeout, **raw_inputs
-                    )
-            else:
-                value = timeout_handler(
-                    self.task.run, timeout=self.task.timeout, **raw_inputs
+
+            # Create a stdout redirect if the task has log_stdout enabled
+            log_context = (
+                redirect_stdout(prefect.utilities.logging.RedirectToLog(self.logger))
+                if getattr(self.task, "log_stdout", False)
+                else nullcontext()
+            )  # type: AbstractContextManager
+
+            with log_context:
+                value = prefect.utilities.executors.run_task_with_timeout(
+                    task=self.task,
+                    args=(),
+                    kwargs=raw_inputs,
+                    logger=self.logger,
                 )
 
         # inform user of timeout
diff --git a/src/prefect/run_configs/__init__.py b/src/prefect/run_configs/__init__.py
index 7d48f6bec1..6b21605053 100644
--- a/src/prefect/run_configs/__init__.py
+++ b/src/prefect/run_configs/__init__.py
@@ -1,2 +1,4 @@
 from .base import RunConfig
 from .kubernetes import KubernetesRun
+from .local import LocalRun
+from .docker import DockerRun
diff --git a/src/prefect/run_configs/docker.py b/src/prefect/run_configs/docker.py
new file mode 100644
index 0000000000..f5a9a67d4d
--- /dev/null
+++ b/src/prefect/run_configs/docker.py
@@ -0,0 +1,40 @@
+from typing import Iterable
+
+from prefect.run_configs.base import RunConfig
+
+
+class DockerRun(RunConfig):
+    """Configure a flow-run to run as a Docker container.
+
+    Note: The functionality here is experimental, and may change between
+    versions without notice. Use at your own risk.
+
+    Args:
+        - image (str, optional): The image to use
+        - env (dict, optional): Additional environment variables to set in the
+            container
+        - labels (Iterable[str], optional): an iterable of labels to apply to this
+            run config. Labels are string identifiers used by Prefect Agents
+            for selecting valid flow runs when polling for work
+
+    Examples:
+
+    Use the defaults set on the agent:
+
+    ```python
+    flow.run_config = DockerRun()
+    ```
+
+    Set an environment variable in the flow run process:
+
+    ```python
+    flow.run_config = DockerRun(env={"SOME_VAR": "value"})
+    ```
+    """
+
+    def __init__(
+        self, *, image: str = None, env: dict = None, labels: Iterable[str] = None
+    ) -> None:
+        super().__init__(labels=labels)
+        self.image = image
+        self.env = env
diff --git a/src/prefect/run_configs/local.py b/src/prefect/run_configs/local.py
new file mode 100644
index 0000000000..0c8dedd8a1
--- /dev/null
+++ b/src/prefect/run_configs/local.py
@@ -0,0 +1,54 @@
+import os
+from typing import Iterable
+
+from prefect.run_configs.base import RunConfig
+
+
+class LocalRun(RunConfig):
+    """Configure a flow-run to run as a Local Process.
+
+    Note: The functionality here is experimental, and may change between
+    versions without notice. Use at your own risk.
+
+    Args:
+        - env (dict, optional): Additional environment variables to set for the process
+        - working_dir (str, optional): Working directory in which to start the
+            process, must already exist. If not provided, will be run in the
+            same directory as the agent.
+        - labels (Iterable[str], optional): an iterable of labels to apply to this
+            run config. Labels are string identifiers used by Prefect Agents
+            for selecting valid flow runs when polling for work
+
+    Examples:
+
+    Use the defaults set on the agent:
+
+    ```python
+    flow.run_config = LocalRun()
+    ```
+
+    Run in a specified working directory:
+
+    ```python
+    flow.run_config = LocalRun(working_dir="/path/to/working-directory")
+    ```
+
+    Set an environment variable in the flow run process:
+
+    ```python
+    flow.run_config = LocalRun(env={"SOME_VAR": "value"})
+    ```
+    """
+
+    def __init__(
+        self,
+        *,
+        env: dict = None,
+        working_dir: str = None,
+        labels: Iterable[str] = None,
+    ) -> None:
+        super().__init__(labels=labels)
+        if working_dir is not None:
+            working_dir = os.path.abspath(working_dir)
+        self.working_dir = working_dir
+        self.env = env
diff --git a/src/prefect/serialization/run_config.py b/src/prefect/serialization/run_config.py
index 9f90d2a62c..611c0ae080 100644
--- a/src/prefect/serialization/run_config.py
+++ b/src/prefect/serialization/run_config.py
@@ -1,7 +1,7 @@
 from marshmallow import fields
 
 from prefect.utilities.serialization import JSONCompatible, OneOfSchema, ObjectSchema
-from prefect.run_configs import KubernetesRun
+from prefect.run_configs import KubernetesRun, LocalRun, DockerRun
 
 
 class RunConfigSchemaBase(ObjectSchema):
@@ -22,7 +22,25 @@ class KubernetesRunSchema(RunConfigSchemaBase):
     memory_request = fields.String(allow_none=True)
 
 
+class LocalRunSchema(RunConfigSchemaBase):
+    class Meta:
+        object_class = LocalRun
+
+    env = fields.Dict(keys=fields.String(), allow_none=True)
+    working_dir = fields.String(allow_none=True)
+
+
+class DockerRunSchema(RunConfigSchemaBase):
+    class Meta:
+        object_class = DockerRun
+
+    image = fields.String(allow_none=True)
+    env = fields.Dict(keys=fields.String(), allow_none=True)
+
+
 class RunConfigSchema(OneOfSchema):
     type_schemas = {
         "KubernetesRun": KubernetesRunSchema,
+        "LocalRun": LocalRunSchema,
+        "DockerRun": DockerRunSchema,
     }
diff --git a/src/prefect/tasks/gcp/storage.py b/src/prefect/tasks/gcp/storage.py
index 72e9f26add..629a617635 100644
--- a/src/prefect/tasks/gcp/storage.py
+++ b/src/prefect/tasks/gcp/storage.py
@@ -1,6 +1,7 @@
 import uuid
 import warnings
 from typing import Union
+import io
 
 from google.cloud.exceptions import NotFound
 
@@ -166,8 +167,7 @@ class GCSDownload(GCSBaseTask):
 
 class GCSUpload(GCSBaseTask):
     """
-    Task template for uploading data to Google Cloud Storage.  Requires the data already
-    be a string.
+    Task template for uploading data to Google Cloud Storage.  Data can be a string or bytes.
 
     Args:
         - bucket (str): default bucket name to upload to
@@ -281,7 +281,7 @@ class GCSUpload(GCSBaseTask):
                 gcs_blob.content_type = content_type
             if content_encoding:
                 gcs_blob.content_encoding = content_encoding
-            gcs_blob.upload_from_file(data)
+            gcs_blob.upload_from_file(io.BytesIO(data))
         return gcs_blob.name
 
 
diff --git a/src/prefect/utilities/compatibility.py b/src/prefect/utilities/compatibility.py
new file mode 100644
index 0000000000..3c49bda767
--- /dev/null
+++ b/src/prefect/utilities/compatibility.py
@@ -0,0 +1,23 @@
+"""
+Functions that exist for compatibility with different python versions. These should
+only be used internally.
+"""
+
+import sys
+
+
+# Provide `nullcontext`
+
+if sys.version_info < (3, 7):
+
+    from contextlib import contextmanager
+    from typing import Iterator
+
+    @contextmanager
+    def nullcontext() -> Iterator[None]:
+        yield
+
+
+else:
+
+    from contextlib import nullcontext  # noqa: F401
diff --git a/src/prefect/utilities/executors.py b/src/prefect/utilities/executors.py
index 2c0027b75c..6f544b305a 100644
--- a/src/prefect/utilities/executors.py
+++ b/src/prefect/utilities/executors.py
@@ -1,4 +1,5 @@
 import copy
+import cloudpickle
 import itertools
 import multiprocessing
 import os
@@ -10,11 +11,11 @@ import warnings
 from concurrent.futures import ThreadPoolExecutor
 from concurrent.futures import TimeoutError as FutureTimeout
 from functools import wraps
-from typing import TYPE_CHECKING, Any, Callable, Dict, List, Union, Sequence
-
-import cloudpickle
+from logging import Logger
+from typing import TYPE_CHECKING, Any, Callable, Dict, List, Union, Sequence, Mapping
 
 import prefect
+from prefect.utilities.logging import get_logger
 
 if TYPE_CHECKING:
     import prefect.engine.runner
@@ -23,6 +24,7 @@ if TYPE_CHECKING:
     from prefect.core.task import Task
     from prefect.engine.state import State
 
+
 StateList = Union["State", List["State"]]
 
 
@@ -77,24 +79,40 @@ def run_with_heartbeat(
     return inner
 
 
-def main_thread_timeout(
-    fn: Callable, *args: Any, timeout: int = None, **kwargs: Any
+def run_with_thread_timeout(
+    fn: Callable,
+    args: Sequence = (),
+    kwargs: Mapping = None,
+    timeout: int = None,
+    logger: Logger = None,
+    name: str = None,
 ) -> Any:
     """
     Helper function for implementing timeouts on function executions.
     Implemented by setting a `signal` alarm on a timer. Must be run in the main thread.
+
     Args:
         - fn (callable): the function to execute
-        - *args (Any): arguments to pass to the function
-        - timeout (int): the length of time to allow for
-            execution before raising a `TimeoutError`, represented as an integer in seconds
-        - **kwargs (Any): keyword arguments to pass to the function
+        - args (Sequence): arguments to pass to the function
+        - kwargs (Mapping): keyword arguments to pass to the function
+        - timeout (int): the length of time to allow for execution before raising a
+            `TimeoutError`, represented as an integer in seconds
+        - logger (Logger): an optional logger to use. If not passed, a logger for the
+            `prefect.executors.run_with_thread_timeout` namespace will be created.
+        - name (str): an optional name to attach to logs for this function run, defaults
+            to the name of the given function. Provides an interface for passing task
+            names for logs.
+
     Returns:
-        - the result of `f(*args, **kwargs)`
+        - the result of `fn(*args, **kwargs)`
+
     Raises:
         - TimeoutError: if function execution exceeds the allowed timeout
         - ValueError: if run from outside the main thread
     """
+    logger = logger or get_logger()
+    name = name or f"Function '{fn.__name__}'"
+    kwargs = kwargs or {}
 
     if timeout is None:
         return fn(*args, **kwargs)
@@ -103,21 +121,25 @@ def main_thread_timeout(
         raise TimeoutError("Execution timed out.")
 
     try:
+        # Set the signal handler for alarms
         signal.signal(signal.SIGALRM, error_handler)
+        # Raise the alarm if `timeout` seconds pass
+        logger.debug(f"{name}: Sending alarm with {timeout}s timeout...")
         signal.alarm(timeout)
+        logger.debug(f"{name}: Executing function in main thread...")
         return fn(*args, **kwargs)
     finally:
         signal.alarm(0)
 
 
-def multiprocessing_safe_retrieve_value(
+def multiprocessing_safe_run_and_retrieve(
     queue: multiprocessing.Queue,
     payload: bytes,
 ) -> None:
     """
     Gets the return value from a function and puts it in a multiprocessing-safe
-    container. Helper function for `multiprocessing_timeout`, must be defined top-level
-    so it can be pickled and sent to `multiprocessing.Process`
+    container. Helper function for `run_with_multiprocess_timeout`, must be defined
+    top-level so it can be pickled and sent to `multiprocessing.Process`
 
     Passing the payload serialized allows us to escape the limitations of the python
     native pickler which will fail on tasks defined in scripts because of name
@@ -131,8 +153,11 @@ def multiprocessing_safe_retrieve_value(
             Expects the following keys:
             - fn (Callable): The function to call
             - args (list): Positional argument values to call the function with
-            - kwargs (dict): Keyword arguments to call the function with
+            - kwargs (Mapping): Keyword arguments to call the function with
             - context (dict): The prefect context dictionary to use during execution
+            - name (str): an optional name to attach to logs for this function run,
+                defaults to the name of the given function. Provides an interface for
+                passing task names for logs.
 
     Returns:
         - None
@@ -141,22 +166,32 @@ def multiprocessing_safe_retrieve_value(
     """
     request = cloudpickle.loads(payload)
 
+    logger = get_logger()
+
     fn: Callable = request["fn"]
     context: dict = request.get("context", {})
     args: Sequence = request.get("args", [])
-    kwargs: dict = request.get("kwargs", {})
+    kwargs: Mapping = request.get("kwargs", {})
+    name: str = request.get("name", f"Function '{fn.__name__}'")
 
     try:
         with prefect.context(context):
+            logger.info(f"{name}: Executing...")
             return_val = fn(*args, **kwargs)
     except Exception as exc:
         return_val = exc
 
+    logger.info(f"{name}: Passing result back to main process...")
     queue.put(cloudpickle.dumps(return_val))
 
 
-def multiprocessing_timeout(
-    fn: Callable, *args: Any, timeout: int = None, **kwargs: Any
+def run_with_multiprocess_timeout(
+    fn: Callable,
+    args: Sequence = (),
+    kwargs: Mapping = None,
+    timeout: int = None,
+    logger: Logger = None,
+    name: str = None,
 ) -> Any:
     """
     Helper function for implementing timeouts on function executions.
@@ -164,10 +199,15 @@ def multiprocessing_timeout(
 
     Args:
         - fn (callable): the function to execute
-        - *args (Any): arguments to pass to the function
+        - args (Sequence): arguments to pass to the function
+        - kwargs (Mapping): keyword arguments to pass to the function
         - timeout (int): the length of time to allow for execution before raising a
             `TimeoutError`, represented as an integer in seconds
-        - **kwargs (Any): keyword arguments to pass to the function
+        - logger (Logger): an optional logger to use. If not passed, a logger for the
+            `prefect.` namespace will be created.
+        - name (str): an optional name to attach to logs for this function run, defaults
+            to the name of the given function. Provides an interface for passing task
+            names for logs.
 
     Returns:
         - the result of `f(*args, **kwargs)`
@@ -176,12 +216,17 @@ def multiprocessing_timeout(
         - AssertionError: if run from a daemonic process
         - TimeoutError: if function execution exceeds the allowed timeout
     """
+    logger = logger or get_logger()
+    name = name or f"Function '{fn.__name__}'"
+    kwargs = kwargs or {}
 
     if timeout is None:
         return fn(*args, **kwargs)
 
+    spawn_mp = multiprocessing.get_context("spawn")
+
     # Create a queue to pass the function return value back
-    queue = multiprocessing.Queue()  # type: multiprocessing.Queue
+    queue = spawn_mp.Queue()  # type: multiprocessing.Queue
 
     # Set internal kwargs for the helper function
     request = {
@@ -189,46 +234,58 @@ def multiprocessing_timeout(
         "args": args,
         "kwargs": kwargs,
         "context": prefect.context.to_dict(),
+        "name": name,
     }
     payload = cloudpickle.dumps(request)
 
-    p = multiprocessing.Process(
-        target=multiprocessing_safe_retrieve_value, args=(queue, payload)
+    run_process = spawn_mp.Process(
+        target=multiprocessing_safe_run_and_retrieve, args=(queue, payload)
     )
-    p.start()
-    p.join(timeout)
-    p.terminate()
+    logger.debug(f"{name}: Sending execution to a new process...")
+    run_process.start()
+    logger.debug(f"{name}: Waiting for process to return with {timeout}s timeout...")
+    run_process.join(timeout)
+    run_process.terminate()
 
     # Handle the process result, if the queue is empty the function did not finish
     # before the timeout
+    logger.debug(f"{name}: Execution process closed, collecting result...")
     if not queue.empty():
-        res = cloudpickle.loads(queue.get())
-        if isinstance(res, Exception):
-            raise res
-        return res
+        result = cloudpickle.loads(queue.get())
+        if isinstance(result, Exception):
+            raise result
+        return result
     else:
-        raise TimeoutError("Execution timed out.")
+        raise TimeoutError(f"Execution timed out for {name}.")
 
 
-def timeout_handler(
-    fn: Callable, *args: Any, timeout: int = None, **kwargs: Any
+def run_task_with_timeout(
+    task: "Task",
+    args: Sequence = (),
+    kwargs: Mapping = None,
+    logger: Logger = None,
 ) -> Any:
     """
-    Helper function for implementing timeouts on function executions.
+    Helper function for implementing timeouts on task executions.
 
     The exact implementation varies depending on whether this function is being
     run in the main thread or a non-daemonic subprocess.  If this is run from a
-    daemonic subprocess or on Windows, the task is run in a
-    `ThreadPoolExecutor` and only a soft timeout is enforced, meaning a
-    `TimeoutError` is raised at the appropriate time but the task continues
-    running in the background.
+    daemonic subprocess or on Windows, the task is run in a `ThreadPoolExecutor`
+    and only a soft timeout is enforced, meaning a `TimeoutError` is raised at the
+    appropriate time but the task continues running in the background.
+
+    The task is passed instead of a function so we can give better logs and messages.
+    If you need to run generic functions with timeout handlers,
+    `run_with_thread_timeout` or `run_with_multiprocess_timeout` can be called directly
 
     Args:
-        - fn (callable): the function to execute
-        - *args (Any): arguments to pass to the function
-        - timeout (int): the length of time to allow for execution before raising a
-            `TimeoutError`, represented as an integer in seconds
-        - **kwargs (Any): keyword arguments to pass to the function
+        - task (Task): the task to execute
+            `task.timeout` specifies the number of seconds to allow `task.run` to run
+            for before terminating
+        - args (Sequence): arguments to pass to the function
+        - kwargs (Mapping): keyword arguments to pass to the function
+        - logger (Logger): an optional logger to use. If not passed, a logger for the
+            `prefect.run_task_with_timeout_handler` namespace will be created.
 
     Returns:
         - the result of `f(*args, **kwargs)`
@@ -236,21 +293,52 @@ def timeout_handler(
     Raises:
         - TimeoutError: if function execution exceeds the allowed timeout
     """
+    logger = logger or get_logger()
+    name = prefect.context.get("task_full_name", task.name)
+    kwargs = kwargs or {}
+
     # if no timeout, just run the function
-    if timeout is None:
-        return fn(*args, **kwargs)
+    if task.timeout is None:
+        return task.run(*args, **kwargs)  # type: ignore
 
     # if we are running the main thread, use a signal to stop execution at the
     # appropriate time; else if we are running in a non-daemonic process, spawn
     # a subprocess to kill at the appropriate time
     if not sys.platform.startswith("win"):
+
         if threading.current_thread() is threading.main_thread():
-            return main_thread_timeout(fn, *args, timeout=timeout, **kwargs)
+            # This case is typically encountered when using a non-parallel or local
+            # multiprocess scheduler because then each worker is in the main
+            # thread
+            logger.debug(f"Task '{name}': Attaching thread based timeout handler...")
+            return run_with_thread_timeout(
+                task.run,
+                args,
+                kwargs,
+                timeout=task.timeout,
+                logger=logger,
+                name=f"Task '{name}'",
+            )
+
         elif multiprocessing.current_process().daemon is False:
-            return multiprocessing_timeout(fn, *args, timeout=timeout, **kwargs)
+            # This case is typically encountered when using a multithread distributed
+            # executor
+            logger.debug(f"Task '{name}': Attaching process based timeout handler...")
+            return run_with_multiprocess_timeout(
+                task.run,
+                args,
+                kwargs,
+                timeout=task.timeout,
+                logger=logger,
+                name=f"Task '{name}'",
+            )
 
+        # We are in a daemonic process and cannot enforce a timeout
+        # This case is typically encountered when using a multiprocess distributed
+        # executor
         soft_timeout_reason = "in a daemonic subprocess"
     else:
+        # We are in windows and cannot enforce a timeout
         soft_timeout_reason = "on Windows"
 
     msg = (
@@ -260,19 +348,22 @@ def timeout_handler(
         "but continue running in the background."
     )
 
+    logger.debug(
+        f"Task '{name}': Falling back to daemonic soft limit timeout handler because "
+        f"we are running {soft_timeout_reason}."
+    )
     warnings.warn(msg, stacklevel=2)
     executor = ThreadPoolExecutor()
 
-    def run_with_ctx(*args: Any, _ctx_dict: dict, **kwargs: Any) -> Any:
-        with prefect.context(_ctx_dict):
-            return fn(*args, **kwargs)
+    def run_with_ctx(context: dict) -> Any:
+        with prefect.context(context):
+            return task.run(*args, **kwargs)  # type: ignore
 
-    fut = executor.submit(
-        run_with_ctx, *args, _ctx_dict=prefect.context.to_dict(), **kwargs
-    )
+    # Run the function in the background and then retrieve its result with a timeout
+    fut = executor.submit(run_with_ctx, prefect.context.to_dict())
 
     try:
-        return fut.result(timeout=timeout)
+        return fut.result(timeout=task.timeout)
     except FutureTimeout as exc:
         raise TimeoutError(
             f"Execution timed out but was executed {soft_timeout_reason} and will "


Test patch:
diff --git a/tests/agent/test_docker_agent.py b/tests/agent/test_docker_agent.py
index f341859d20..15a20f7717 100644
--- a/tests/agent/test_docker_agent.py
+++ b/tests/agent/test_docker_agent.py
@@ -7,76 +7,74 @@ from prefect import context
 from prefect.agent.docker.agent import DockerAgent, _stream_container_logs
 from prefect.environments import LocalEnvironment
 from prefect.environments.storage import Docker, Local
+from prefect.run_configs import DockerRun, LocalRun
 from prefect.utilities.configuration import set_temporary_config
 from prefect.utilities.graphql import GraphQLResult
 
+docker = pytest.importorskip("docker")
 
-def test_docker_agent_init(monkeypatch, cloud_api):
-    api = MagicMock()
+
+@pytest.fixture(autouse=True)
+def mock_cloud_config(cloud_api):
+    with set_temporary_config(
+        {"cloud.agent.auth_token": "TEST_TOKEN", "logging.log_to_cloud": True}
+    ):
+        yield
+
+
+@pytest.fixture
+def api(monkeypatch):
+    client = MagicMock()
+    client.ping.return_value = True
+    client.create_container.return_value = {"Id": "container_id"}
+    client.create_host_config.return_value = {"AutoRemove": True}
     monkeypatch.setattr(
         "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
+        MagicMock(return_value=client),
     )
+    return client
 
+
+def test_docker_agent_init(api):
     agent = DockerAgent()
     assert agent
-    assert agent.agent_config_id == None
+    assert agent.agent_config_id is None
     assert agent.labels == []
     assert agent.name == "agent"
 
 
-def test_docker_agent_config_options(monkeypatch, cloud_api):
-    import docker  # DockerAgent imports docker within the constructor
-
-    api = MagicMock()
-    monkeypatch.setattr("docker.APIClient", api)
-    monkeypatch.setattr("prefect.agent.docker.agent.platform", "osx")
-
-    with set_temporary_config({"cloud.agent.auth_token": "TEST_TOKEN"}):
-        agent = DockerAgent(name="test")
-        assert agent.name == "test"
-        assert agent.client.get_auth_token() == "TEST_TOKEN"
-        assert agent.logger
-        assert not agent.no_pull
-        assert api.call_args[1]["base_url"] == "unix://var/run/docker.sock"
-
-
-def test_docker_agent_daemon_url_responds_to_system(monkeypatch, cloud_api):
-    import docker  # DockerAgent imports docker within the constructor
-
+@pytest.mark.parametrize(
+    "platform, url",
+    [
+        ("osx", "unix://var/run/docker.sock"),
+        ("win32", "npipe:////./pipe/docker_engine"),
+    ],
+)
+def test_docker_agent_config_options(platform, url, monkeypatch):
     api = MagicMock()
     monkeypatch.setattr("docker.APIClient", api)
-    monkeypatch.setattr("prefect.agent.docker.agent.platform", "win32")
-
-    with set_temporary_config({"cloud.agent.auth_token": "TEST_TOKEN"}):
-        agent = DockerAgent()
-        assert agent.client.get_auth_token() == "TEST_TOKEN"
-        assert agent.logger
-        assert not agent.no_pull
-        assert api.call_args[1]["base_url"] == "npipe:////./pipe/docker_engine"
+    monkeypatch.setattr("prefect.agent.docker.agent.platform", platform)
 
+    agent = DockerAgent(name="test")
+    assert agent.name == "test"
+    assert agent.client.get_auth_token() == "TEST_TOKEN"
+    assert agent.logger
+    assert not agent.no_pull
+    assert api.call_args[1]["base_url"] == url
 
-def test_docker_agent_config_options_populated(monkeypatch, cloud_api):
-    import docker  # DockerAgent imports docker within the constructor
 
+def test_docker_agent_config_options_populated(monkeypatch):
     api = MagicMock()
     monkeypatch.setattr("docker.APIClient", api)
 
-    with set_temporary_config({"cloud.agent.auth_token": "TEST_TOKEN"}):
-        agent = DockerAgent(base_url="url", no_pull=True)
-        assert agent.client.get_auth_token() == "TEST_TOKEN"
-        assert agent.logger
-        assert agent.no_pull
-        assert api.call_args[1]["base_url"] == "url"
-
+    agent = DockerAgent(base_url="url", no_pull=True)
+    assert agent.client.get_auth_token() == "TEST_TOKEN"
+    assert agent.logger
+    assert agent.no_pull
+    assert api.call_args[1]["base_url"] == "url"
 
-def test_docker_agent_no_pull(monkeypatch, cloud_api):
-    api = MagicMock()
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
 
+def test_docker_agent_no_pull(api):
     agent = DockerAgent()
     assert not agent.no_pull
 
@@ -96,143 +94,90 @@ def test_docker_agent_no_pull(monkeypatch, cloud_api):
         assert not agent.no_pull
 
 
-def test_docker_agent_ping(monkeypatch, cloud_api):
-    api = MagicMock()
-    api.ping.return_value = True
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
-
-    agent = DockerAgent()
+def test_docker_agent_ping(api):
+    DockerAgent()
     assert api.ping.called
 
 
-def test_docker_agent_ping_exception(monkeypatch, cloud_api):
-    api = MagicMock()
-    api.ping.return_value = True
+def test_docker_agent_ping_exception(api):
     api.ping.side_effect = Exception()
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
 
     with pytest.raises(Exception):
-        agent = DockerAgent()
-
+        DockerAgent()
 
-def test_populate_env_vars_uses_user_provided_env_vars(monkeypatch, cloud_api):
-    api = MagicMock()
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
 
-    with set_temporary_config(
-        {
-            "cloud.agent.auth_token": "token",
-            "cloud.api": "api",
-            "logging.log_to_cloud": True,
-        }
-    ):
-        agent = DockerAgent(env_vars=dict(AUTH_THING="foo"))
+def test_populate_env_vars_from_agent_config(api):
+    agent = DockerAgent(env_vars=dict(AUTH_THING="foo"))
 
-        env_vars = agent.populate_env_vars(
-            GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
-        )
+    env_vars = agent.populate_env_vars(
+        GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
+    )
 
     assert env_vars["AUTH_THING"] == "foo"
 
 
-def test_populate_env_vars(monkeypatch, cloud_api):
-    api = MagicMock()
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
+def test_populate_env_vars(api):
+    agent = DockerAgent()
 
-    with set_temporary_config(
-        {
-            "cloud.agent.auth_token": "token",
-            "cloud.api": "api",
-            "logging.log_to_cloud": True,
-        }
-    ):
-        agent = DockerAgent()
+    env_vars = agent.populate_env_vars(
+        GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
+    )
 
-        env_vars = agent.populate_env_vars(
-            GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
-        )
+    expected_vars = {
+        "PREFECT__CLOUD__API": "https://api.prefect.io",
+        "PREFECT__CLOUD__AUTH_TOKEN": "TEST_TOKEN",
+        "PREFECT__CLOUD__AGENT__LABELS": "[]",
+        "PREFECT__CONTEXT__FLOW_RUN_ID": "id",
+        "PREFECT__CONTEXT__FLOW_ID": "foo",
+        "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
+        "PREFECT__LOGGING__LOG_TO_CLOUD": "true",
+        "PREFECT__LOGGING__LEVEL": "INFO",
+        "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
+        "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
+    }
 
-        expected_vars = {
-            "PREFECT__CLOUD__API": "api",
-            "PREFECT__CLOUD__AUTH_TOKEN": "token",
-            "PREFECT__CLOUD__AGENT__LABELS": "[]",
-            "PREFECT__CONTEXT__FLOW_RUN_ID": "id",
-            "PREFECT__CONTEXT__FLOW_ID": "foo",
-            "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
-            "PREFECT__LOGGING__LOG_TO_CLOUD": "true",
-            "PREFECT__LOGGING__LEVEL": "INFO",
-            "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
-            "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
-        }
+    assert env_vars == expected_vars
 
-        assert env_vars == expected_vars
 
+def test_populate_env_vars_includes_agent_labels(api):
+    agent = DockerAgent(labels=["42", "marvin"])
 
-def test_populate_env_vars_includes_agent_labels(monkeypatch, cloud_api):
-    api = MagicMock()
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
+    env_vars = agent.populate_env_vars(
+        GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
     )
+    assert env_vars["PREFECT__CLOUD__AGENT__LABELS"] == "['42', 'marvin']"
 
-    with set_temporary_config(
-        {
-            "cloud.agent.auth_token": "token",
-            "cloud.api": "api",
-            "logging.log_to_cloud": True,
-        }
-    ):
-        agent = DockerAgent(labels=["42", "marvin"])
 
-        env_vars = agent.populate_env_vars(
-            GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
-        )
+@pytest.mark.parametrize("flag", [True, False])
+def test_populate_env_vars_sets_log_to_cloud(flag, api):
+    agent = DockerAgent(labels=["42", "marvin"], no_cloud_logs=flag)
 
-        expected_vars = {
-            "PREFECT__CLOUD__API": "api",
-            "PREFECT__CLOUD__AGENT__LABELS": "['42', 'marvin']",
-            "PREFECT__CLOUD__AUTH_TOKEN": "token",
-            "PREFECT__CONTEXT__FLOW_RUN_ID": "id",
-            "PREFECT__CONTEXT__FLOW_ID": "foo",
-            "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
-            "PREFECT__LOGGING__LOG_TO_CLOUD": "true",
-            "PREFECT__LOGGING__LEVEL": "INFO",
-            "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
-            "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
-        }
+    env_vars = agent.populate_env_vars(
+        GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
+    )
+    assert env_vars["PREFECT__LOGGING__LOG_TO_CLOUD"] == str(not flag).lower()
 
-        assert env_vars == expected_vars
 
+def test_populate_env_vars_from_run_config(api):
+    agent = DockerAgent(env_vars={"KEY1": "VAL1", "KEY2": "VAL2"})
 
-@pytest.mark.parametrize("flag", [True, False])
-def test_populate_env_vars_is_responsive_to_logging_config(
-    monkeypatch, cloud_api, flag
-):
-    api = MagicMock()
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
+    run = DockerRun(
+        env={"KEY2": "OVERRIDE", "PREFECT__LOGGING__LEVEL": "TEST"},
     )
 
-    with set_temporary_config({"cloud.agent.auth_token": "token", "cloud.api": "api"}):
-        agent = DockerAgent(labels=["42", "marvin"], no_cloud_logs=flag)
-
-        env_vars = agent.populate_env_vars(
-            GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
-        )
-    assert env_vars["PREFECT__LOGGING__LOG_TO_CLOUD"] == str(not flag).lower()
+    env_vars = agent.populate_env_vars(
+        GraphQLResult(
+            {
+                "id": "id",
+                "name": "name",
+                "flow": {"id": "foo", "run_config": run.serialize()},
+            }
+        ),
+        run,
+    )
+    assert env_vars["KEY1"] == "VAL1"
+    assert env_vars["KEY2"] == "OVERRIDE"
+    assert env_vars["PREFECT__LOGGING__LEVEL"] == "TEST"
 
 
 @pytest.mark.parametrize(
@@ -244,15 +189,7 @@ def test_populate_env_vars_is_responsive_to_logging_config(
         ("0.13.1+134", "prefect execute flow-run"),
     ],
 )
-def test_docker_agent_deploy_flow(core_version, command, monkeypatch, cloud_api):
-    api = MagicMock()
-    api.ping.return_value = True
-    api.create_container.return_value = {"Id": "container_id"}
-    api.create_host_config.return_value = {"AutoRemove": True}
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
+def test_docker_agent_deploy_flow(core_version, command, api):
 
     agent = DockerAgent()
     agent.deploy_flow(
@@ -284,16 +221,7 @@ def test_docker_agent_deploy_flow(core_version, command, monkeypatch, cloud_api)
     assert api.start.call_args[1]["container"] == "container_id"
 
 
-def test_docker_agent_deploy_flow_uses_environment_metadata(monkeypatch, cloud_api):
-    api = MagicMock()
-    api.ping.return_value = True
-    api.create_container.return_value = {"Id": "container_id"}
-    api.create_host_config.return_value = {"AutoRemove": True}
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
-
+def test_docker_agent_deploy_flow_uses_environment_metadata(api):
     agent = DockerAgent()
     agent.deploy_flow(
         flow_run=GraphQLResult(
@@ -324,28 +252,54 @@ def test_docker_agent_deploy_flow_uses_environment_metadata(monkeypatch, cloud_a
     assert api.start.call_args[1]["container"] == "container_id"
 
 
-def test_docker_agent_deploy_flow_storage_raises(monkeypatch, cloud_api):
+@pytest.mark.parametrize("image_on_run_config", [True, False])
+def test_docker_agent_deploy_flow_run_config(api, image_on_run_config):
+    if image_on_run_config:
+        storage = Local()
+        image = "on-run-config"
+        run = DockerRun(image=image, env={"TESTING": "VALUE"})
+    else:
+        storage = Docker(
+            registry_url="testing", image_name="on-storage", image_tag="tag"
+        )
+        image = "testing/on-storage:tag"
+        run = DockerRun(env={"TESTING": "VALUE"})
 
-    monkeypatch.setattr("prefect.agent.agent.Client", MagicMock())
-    api = MagicMock()
-    api.ping.return_value = True
-    api.create_container.return_value = {"Id": "container_id"}
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
+    agent = DockerAgent()
+    agent.deploy_flow(
+        flow_run=GraphQLResult(
+            {
+                "flow": GraphQLResult(
+                    {
+                        "id": "foo",
+                        "storage": storage.serialize(),
+                        "run_config": run.serialize(),
+                        "core_version": "0.13.11",
+                    }
+                ),
+                "id": "id",
+                "name": "name",
+            }
+        )
     )
 
+    assert api.create_container.called
+    assert api.create_container.call_args[0][0] == image
+    assert api.create_container.call_args[1]["environment"]["TESTING"] == "VALUE"
+
+
+def test_docker_agent_deploy_flow_unsupported_run_config(api):
     agent = DockerAgent()
 
-    with pytest.raises(ValueError):
+    with pytest.raises(TypeError, match="Unsupported RunConfig type: LocalRun"):
         agent.deploy_flow(
             flow_run=GraphQLResult(
                 {
                     "flow": GraphQLResult(
                         {
                             "storage": Local().serialize(),
+                            "run_config": LocalRun().serialize(),
                             "id": "foo",
-                            "environment": LocalEnvironment().serialize(),
                             "core_version": "0.13.0",
                         }
                     ),
@@ -359,16 +313,34 @@ def test_docker_agent_deploy_flow_storage_raises(monkeypatch, cloud_api):
     assert not api.pull.called
 
 
-def test_docker_agent_deploy_flow_no_pull(monkeypatch, cloud_api):
+def test_docker_agent_deploy_flow_storage_raises(monkeypatch, api):
+    monkeypatch.setattr("prefect.agent.agent.Client", MagicMock())
+
+    agent = DockerAgent()
+
+    with pytest.raises(ValueError):
+        agent.deploy_flow(
+            flow_run=GraphQLResult(
+                {
+                    "flow": GraphQLResult(
+                        {
+                            "storage": Local().serialize(),
+                            "id": "foo",
+                            "environment": LocalEnvironment().serialize(),
+                            "core_version": "0.13.0",
+                        }
+                    ),
+                    "id": "id",
+                    "name": "name",
+                    "version": "version",
+                }
+            )
+        )
+
+    assert not api.pull.called
 
-    api = MagicMock()
-    api.ping.return_value = True
-    api.create_container.return_value = {"Id": "container_id"}
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
 
+def test_docker_agent_deploy_flow_no_pull(api):
     agent = DockerAgent(no_pull=True)
     agent.deploy_flow(
         flow_run=GraphQLResult(
@@ -394,18 +366,7 @@ def test_docker_agent_deploy_flow_no_pull(monkeypatch, cloud_api):
     assert api.start.called
 
 
-def test_docker_agent_deploy_flow_no_pull_using_environment_metadata(
-    monkeypatch, cloud_api
-):
-
-    api = MagicMock()
-    api.ping.return_value = True
-    api.create_container.return_value = {"Id": "container_id"}
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
-
+def test_docker_agent_deploy_flow_no_pull_using_environment_metadata(api):
     agent = DockerAgent(no_pull=True)
     agent.deploy_flow(
         flow_run=GraphQLResult(
@@ -431,15 +392,7 @@ def test_docker_agent_deploy_flow_no_pull_using_environment_metadata(
     assert api.start.called
 
 
-def test_docker_agent_deploy_flow_reg_allow_list_allowed(monkeypatch, cloud_api):
-    api = MagicMock()
-    api.ping.return_value = True
-    api.create_container.return_value = {"Id": "container_id"}
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
-
+def test_docker_agent_deploy_flow_reg_allow_list_allowed(api):
     agent = DockerAgent(reg_allow_list=["test1"])
 
     agent.deploy_flow(
@@ -466,15 +419,7 @@ def test_docker_agent_deploy_flow_reg_allow_list_allowed(monkeypatch, cloud_api)
     assert api.start.called
 
 
-def test_docker_agent_deploy_flow_reg_allow_list_not_allowed(monkeypatch, cloud_api):
-    api = MagicMock()
-    api.ping.return_value = True
-    api.create_container.return_value = {"Id": "container_id"}
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
-
+def test_docker_agent_deploy_flow_reg_allow_list_not_allowed(api):
     agent = DockerAgent(reg_allow_list=["test1"])
 
     with pytest.raises(ValueError) as error:
@@ -508,19 +453,10 @@ def test_docker_agent_deploy_flow_reg_allow_list_not_allowed(monkeypatch, cloud_
     assert str(error.value) == expected_error
 
 
-def test_docker_agent_deploy_flow_show_flow_logs(monkeypatch, cloud_api):
-
+def test_docker_agent_deploy_flow_show_flow_logs(api, monkeypatch):
     process = MagicMock()
     monkeypatch.setattr("multiprocessing.Process", process)
 
-    api = MagicMock()
-    api.ping.return_value = True
-    api.create_container.return_value = {"Id": "container_id"}
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
-
     agent = DockerAgent(show_flow_logs=True)
     agent.deploy_flow(
         flow_run=GraphQLResult(
@@ -554,15 +490,8 @@ def test_docker_agent_deploy_flow_show_flow_logs(monkeypatch, cloud_api):
     assert api.start.called
 
 
-def test_docker_agent_shutdown_terminates_child_processes(monkeypatch, cloud_api):
+def test_docker_agent_shutdown_terminates_child_processes(monkeypatch, api):
     monkeypatch.setattr("prefect.agent.agent.Client", MagicMock())
-    api = MagicMock()
-    api.ping.return_value = True
-    api.create_container.return_value = {"Id": "container_id"}
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
 
     proc = MagicMock(is_alive=MagicMock(return_value=True))
     agent = DockerAgent(show_flow_logs=True)
@@ -573,16 +502,7 @@ def test_docker_agent_shutdown_terminates_child_processes(monkeypatch, cloud_api
     assert proc.terminate.called
 
 
-def test_docker_agent_deploy_flow_no_registry_does_not_pull(monkeypatch, cloud_api):
-
-    api = MagicMock()
-    api.ping.return_value = True
-    api.create_container.return_value = {"Id": "container_id"}
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
-
+def test_docker_agent_deploy_flow_no_registry_does_not_pull(api):
     agent = DockerAgent()
     agent.deploy_flow(
         flow_run=GraphQLResult(
@@ -608,48 +528,24 @@ def test_docker_agent_deploy_flow_no_registry_does_not_pull(monkeypatch, cloud_a
     assert api.start.called
 
 
-def test_docker_agent_heartbeat_gocase(monkeypatch, cloud_api):
-    api = MagicMock()
-    api.ping.return_value = True
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
-
+def test_docker_agent_heartbeat_gocase(api):
     agent = DockerAgent()
     agent.heartbeat()
     assert api.ping.call_count == 2
 
 
-def test_docker_agent_heartbeat_exits_on_failure(monkeypatch, cloud_api, caplog):
-    api = MagicMock()
-    api.ping.return_value = True
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
-
+def test_docker_agent_heartbeat_exits_on_failure(api, caplog):
     agent = DockerAgent()
     api.ping.return_value = False
-    agent.heartbeat()
-    agent.heartbeat()
-    agent.heartbeat()
-    agent.heartbeat()
-    agent.heartbeat()
+    for _ in range(5):
+        agent.heartbeat()
     with pytest.raises(SystemExit):
         agent.heartbeat()
     assert "Cannot reconnect to Docker daemon. Agent is shutting down." in caplog.text
     assert api.ping.call_count == 7
 
 
-def test_docker_agent_heartbeat_logs_reconnect(monkeypatch, cloud_api, caplog):
-    api = MagicMock()
-    api.ping.return_value = True
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
-
+def test_docker_agent_heartbeat_logs_reconnect(api, caplog):
     agent = DockerAgent()
     api.ping.return_value = False
     agent.heartbeat()
@@ -660,14 +556,7 @@ def test_docker_agent_heartbeat_logs_reconnect(monkeypatch, cloud_api, caplog):
     assert "Reconnected to Docker daemon" in caplog.text
 
 
-def test_docker_agent_heartbeat_resets_fail_count(monkeypatch, cloud_api, caplog):
-    api = MagicMock()
-    api.ping.return_value = True
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
-
+def test_docker_agent_heartbeat_resets_fail_count(api, caplog):
     agent = DockerAgent()
     api.ping.return_value = False
     agent.heartbeat()
@@ -679,13 +568,7 @@ def test_docker_agent_heartbeat_resets_fail_count(monkeypatch, cloud_api, caplog
     assert api.ping.call_count == 4
 
 
-def test_docker_agent_init_volume_empty_options(monkeypatch, cloud_api):
-    api = MagicMock()
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
-
+def test_docker_agent_init_volume_empty_options(api):
     agent = DockerAgent()
     assert agent
     assert agent.named_volumes == []
@@ -705,12 +588,7 @@ def test_docker_agent_init_volume_empty_options(monkeypatch, cloud_api):
         ("\n../some/path", True),  # it is up to the caller to strip the string
     ],
 )
-def test_docker_agent_is_named_volume_unix(monkeypatch, cloud_api, path, result):
-    api = MagicMock()
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
+def test_docker_agent_is_named_volume_unix(monkeypatch, api, path, result):
     monkeypatch.setattr("prefect.agent.docker.agent.platform", "osx")
 
     agent = DockerAgent()
@@ -728,12 +606,7 @@ def test_docker_agent_is_named_volume_unix(monkeypatch, cloud_api, path, result)
         ("\\\\\\some\\path", False),
     ],
 )
-def test_docker_agent_is_named_volume_win32(monkeypatch, cloud_api, path, result):
-    api = MagicMock()
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
+def test_docker_agent_is_named_volume_win32(monkeypatch, api, path, result):
     monkeypatch.setattr("prefect.agent.docker.agent.platform", "win32")
 
     agent = DockerAgent()
@@ -796,14 +669,8 @@ def test_docker_agent_is_named_volume_win32(monkeypatch, cloud_api, path, result
     ],
 )
 def test_docker_agent_parse_volume_spec_unix(
-    monkeypatch, cloud_api, candidate, named_volumes, container_mount_paths, host_spec
+    api, candidate, named_volumes, container_mount_paths, host_spec
 ):
-    api = MagicMock()
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
-
     agent = DockerAgent()
 
     (
@@ -865,14 +732,8 @@ def test_docker_agent_parse_volume_spec_unix(
     ],
 )
 def test_docker_agent_parse_volume_spec_win(
-    monkeypatch, cloud_api, candidate, named_volumes, container_mount_paths, host_spec
+    api, candidate, named_volumes, container_mount_paths, host_spec
 ):
-    api = MagicMock()
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
-
     agent = DockerAgent()
 
     (
@@ -896,55 +757,16 @@ def test_docker_agent_parse_volume_spec_win(
     ],
 )
 def test_docker_agent_parse_volume_spec_raises_on_invalid_spec(
-    monkeypatch, cloud_api, candidate, exception_type
+    api, candidate, exception_type
 ):
-    api = MagicMock()
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
-
     agent = DockerAgent()
 
     with pytest.raises(exception_type):
         agent._parse_volume_spec([candidate])
 
 
-def test_docker_agent_start_max_polls(monkeypatch, runner_token, cloud_api):
-    api = MagicMock()
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
-
-    on_shutdown = MagicMock()
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent.on_shutdown", on_shutdown
-    )
-
-    agent_process = MagicMock()
-    monkeypatch.setattr("prefect.agent.agent.Agent.agent_process", agent_process)
-
-    agent_connect = MagicMock(return_value="id")
-    monkeypatch.setattr("prefect.agent.agent.Agent.agent_connect", agent_connect)
-
-    heartbeat = MagicMock()
-    monkeypatch.setattr("prefect.agent.docker.agent.DockerAgent.heartbeat", heartbeat)
-
-    agent = DockerAgent(max_polls=1)
-    agent.start()
-
-    assert agent_process.called
-    assert heartbeat.called
-
-
-def test_docker_agent_start_max_polls_count(monkeypatch, runner_token, cloud_api):
-    api = MagicMock()
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
-
+@pytest.mark.parametrize("max_polls", [0, 1, 2])
+def test_docker_agent_start_max_polls(max_polls, api, monkeypatch, runner_token):
     on_shutdown = MagicMock()
     monkeypatch.setattr(
         "prefect.agent.docker.agent.DockerAgent.on_shutdown", on_shutdown
@@ -959,53 +781,17 @@ def test_docker_agent_start_max_polls_count(monkeypatch, runner_token, cloud_api
     heartbeat = MagicMock()
     monkeypatch.setattr("prefect.agent.docker.agent.DockerAgent.heartbeat", heartbeat)
 
-    agent = DockerAgent(max_polls=2)
+    agent = DockerAgent(max_polls=max_polls)
     agent.start()
 
-    assert on_shutdown.call_count == 1
-    assert agent_process.call_count == 2
+    assert agent_connect.call_count == 1
+    assert agent_process.call_count == max_polls
     assert heartbeat.call_count == 1
-
-
-def test_docker_agent_start_max_polls_zero(monkeypatch, runner_token, cloud_api):
-    api = MagicMock()
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
-
-    on_shutdown = MagicMock()
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent.on_shutdown", on_shutdown
-    )
-
-    agent_process = MagicMock()
-    monkeypatch.setattr("prefect.agent.agent.Agent.agent_process", agent_process)
-
-    agent_connect = MagicMock(return_value="id")
-    monkeypatch.setattr("prefect.agent.agent.Agent.agent_connect", agent_connect)
-
-    heartbeat = MagicMock()
-    monkeypatch.setattr("prefect.agent.docker.agent.DockerAgent.heartbeat", heartbeat)
-
-    agent = DockerAgent(max_polls=0)
-    agent.start()
-
     assert on_shutdown.call_count == 1
-    assert agent_process.call_count == 0
-    assert heartbeat.call_count == 1
 
 
-def test_docker_agent_network(monkeypatch, cloud_api):
-
-    api = MagicMock()
-    api.ping.return_value = True
-    api.create_container.return_value = {"Id": "container_id"}
+def test_docker_agent_network(api):
     api.create_networking_config.return_value = {"test-network": "config"}
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
 
     agent = DockerAgent(network="test-network")
     agent.deploy_flow(
@@ -1033,17 +819,8 @@ def test_docker_agent_network(monkeypatch, cloud_api):
 
 
 def test_docker_agent_deploy_with_interface_check_linux(
-    monkeypatch, cloud_api, linux_platform
+    api, monkeypatch, linux_platform
 ):
-
-    api = MagicMock()
-    api.ping.return_value = True
-    api.create_container.return_value = {"Id": "container_id"}
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
-
     get_ip = MagicMock()
     monkeypatch.setattr("prefect.agent.docker.agent.get_docker_ip", get_ip)
 
@@ -1071,17 +848,8 @@ def test_docker_agent_deploy_with_interface_check_linux(
 
 
 def test_docker_agent_deploy_with_no_interface_check_linux(
-    monkeypatch, cloud_api, linux_platform
+    api, monkeypatch, linux_platform
 ):
-
-    api = MagicMock()
-    api.ping.return_value = True
-    api.create_container.return_value = {"Id": "container_id"}
-    monkeypatch.setattr(
-        "prefect.agent.docker.agent.DockerAgent._get_docker_client",
-        MagicMock(return_value=api),
-    )
-
     get_ip = MagicMock()
     monkeypatch.setattr("prefect.agent.docker.agent.get_docker_ip", get_ip)
 
diff --git a/tests/agent/test_local_agent.py b/tests/agent/test_local_agent.py
index 3c8fba344a..4c2d4d657d 100644
--- a/tests/agent/test_local_agent.py
+++ b/tests/agent/test_local_agent.py
@@ -1,3 +1,4 @@
+import os
 import socket
 from unittest.mock import MagicMock
 
@@ -8,109 +9,77 @@ from testfixtures import compare, LogCapture
 
 from prefect.agent.local import LocalAgent
 from prefect.environments.storage import Docker, Local, Azure, GCS, S3, Webhook, GitLab
+from prefect.run_configs import LocalRun, KubernetesRun
 from prefect.utilities.configuration import set_temporary_config
 from prefect.utilities.graphql import GraphQLResult
 
+DEFAULT_AGENT_LABELS = [
+    socket.gethostname(),
+    "azure-flow-storage",
+    "gcs-flow-storage",
+    "s3-flow-storage",
+    "github-flow-storage",
+    "webhook-flow-storage",
+    "gitlab-flow-storage",
+]
 
-def test_local_agent_init(cloud_api):
+
+@pytest.fixture(autouse=True)
+def mock_cloud_config(cloud_api):
+    with set_temporary_config(
+        {"cloud.agent.auth_token": "TEST_TOKEN", "logging.log_to_cloud": True}
+    ):
+        yield
+
+
+def test_local_agent_init():
     agent = LocalAgent()
-    assert agent
-    assert agent.agent_config_id == None
-    assert set(agent.labels) == {
-        socket.gethostname(),
-        "azure-flow-storage",
-        "s3-flow-storage",
-        "gcs-flow-storage",
-        "github-flow-storage",
-        "webhook-flow-storage",
-        "gitlab-flow-storage",
-    }
+    assert agent.agent_config_id is None
+    assert set(agent.labels) == set(DEFAULT_AGENT_LABELS)
     assert agent.name == "agent"
 
 
-def test_local_agent_deduplicates_labels(cloud_api):
+def test_local_agent_deduplicates_labels():
     agent = LocalAgent(labels=["azure-flow-storage"])
-    assert agent
+    assert sorted(agent.labels) == sorted(DEFAULT_AGENT_LABELS)
+
+
+def test_local_agent_config_options():
+    agent = LocalAgent(
+        name="test",
+        labels=["test_label"],
+        import_paths=["test_path"],
+    )
+    assert agent.name == "test"
+    assert agent.client.get_auth_token() == "TEST_TOKEN"
+    assert agent.logger
+    assert agent.log_to_cloud is True
+    assert agent.processes == set()
+    assert agent.import_paths == ["test_path"]
+    assert set(agent.labels) == {"test_label", *DEFAULT_AGENT_LABELS}
+
+
+def test_local_agent_config_no_storage_labels():
+    agent = LocalAgent(
+        labels=["test_label"],
+        storage_labels=False,
+    )
     assert set(agent.labels) == {
         socket.gethostname(),
-        "azure-flow-storage",
-        "s3-flow-storage",
-        "gcs-flow-storage",
-        "github-flow-storage",
-        "webhook-flow-storage",
-        "gitlab-flow-storage",
+        "test_label",
     }
-    assert len(agent.labels) == len(set(agent.labels))
 
 
-def test_local_agent_config_options(cloud_api):
-    with set_temporary_config(
-        {"cloud.agent.auth_token": "TEST_TOKEN", "logging.log_to_cloud": True}
-    ):
-        agent = LocalAgent(
-            name="test",
-            labels=["test_label"],
-            import_paths=["test_path"],
-            hostname_label=False,
-        )
-        assert agent.name == "test"
-        assert agent.client.get_auth_token() == "TEST_TOKEN"
-        assert agent.logger
-        assert agent.log_to_cloud is True
-        assert agent.processes == set()
-        assert agent.import_paths == ["test_path"]
-        assert set(agent.labels) == {
-            "azure-flow-storage",
-            "s3-flow-storage",
-            "gcs-flow-storage",
-            "github-flow-storage",
-            "webhook-flow-storage",
-            "gitlab-flow-storage",
-            "test_label",
-        }
+@pytest.mark.parametrize("hostname_label", [True, False])
+def test_local_agent_config_options_hostname(hostname_label):
+    agent = LocalAgent(labels=["test_label"], hostname_label=hostname_label)
+    expected = {"test_label", *DEFAULT_AGENT_LABELS}
+    if not hostname_label:
+        expected.discard(socket.gethostname())
+    assert set(agent.labels) == expected
 
 
-def test_local_agent_config_no_storage_labels(cloud_api):
-    with set_temporary_config(
-        {"cloud.agent.auth_token": "TEST_TOKEN", "logging.log_to_cloud": True}
-    ):
-        agent = LocalAgent(
-            labels=["test_label"],
-            storage_labels=False,
-        )
-        assert set(agent.labels) == {
-            socket.gethostname(),
-            "test_label",
-        }
-
-
-@pytest.mark.parametrize("flag", [True, False])
-def test_local_agent_responds_to_logging_config(cloud_api, flag):
-    with set_temporary_config({"cloud.agent.auth_token": "TEST_TOKEN"}):
-        agent = LocalAgent(no_cloud_logs=flag)
-        assert agent.log_to_cloud is not flag
-        env_vars = agent.populate_env_vars(
-            GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
-        )
-        assert env_vars["PREFECT__LOGGING__LOG_TO_CLOUD"] == str(not flag).lower()
-
-
-def test_local_agent_config_options_hostname(cloud_api):
-    with set_temporary_config({"cloud.agent.auth_token": "TEST_TOKEN"}):
-        agent = LocalAgent(labels=["test_label"])
-        assert set(agent.labels) == {
-            "test_label",
-            socket.gethostname(),
-            "azure-flow-storage",
-            "s3-flow-storage",
-            "gcs-flow-storage",
-            "github-flow-storage",
-            "webhook-flow-storage",
-            "gitlab-flow-storage",
-        }
-
-
-def test_local_agent_uses_ip_if_dockerdesktop_hostname(monkeypatch, cloud_api):
+def test_local_agent_uses_ip_if_dockerdesktop_hostname(monkeypatch):
     monkeypatch.setattr("socket.gethostname", MagicMock(return_value="docker-desktop"))
     monkeypatch.setattr("socket.gethostbyname", MagicMock(return_value="IP"))
 
@@ -118,68 +87,21 @@ def test_local_agent_uses_ip_if_dockerdesktop_hostname(monkeypatch, cloud_api):
     assert "IP" in agent.labels
 
 
-def test_populate_env_vars_uses_user_provided_env_vars(cloud_api):
-    with set_temporary_config(
-        {
-            "cloud.api": "api",
-            "logging.log_to_cloud": True,
-            "cloud.agent.auth_token": "token",
-        }
-    ):
-        agent = LocalAgent(env_vars=dict(AUTH_THING="foo"))
-
-        env_vars = agent.populate_env_vars(
-            GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
-        )
-
-    assert env_vars["AUTH_THING"] == "foo"
-
-
-def test_populate_env_vars_uses_user_provided_env_vars_removes_nones(cloud_api):
-    with set_temporary_config(
-        {
-            "cloud.api": "api",
-            "logging.log_to_cloud": True,
-            "cloud.agent.auth_token": "token",
-        }
-    ):
-        agent = LocalAgent(env_vars=dict(MISSING_VAR=None))
-
-        env_vars = agent.populate_env_vars(
-            GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
-        )
-
-    assert "MISSING_VAR" not in env_vars
+def test_populate_env_vars():
+    agent = LocalAgent()
 
+    env_vars = agent.populate_env_vars(
+        GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
+    )
+    python_path = env_vars.pop("PYTHONPATH", "")
+    assert os.getcwd() in python_path
 
-def test_populate_env_vars(cloud_api):
-    with set_temporary_config(
+    expected = os.environ.copy()
+    expected.update(
         {
-            "cloud.api": "api",
-            "logging.log_to_cloud": True,
-            "cloud.agent.auth_token": "token",
-        }
-    ):
-        agent = LocalAgent()
-
-        env_vars = agent.populate_env_vars(
-            GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
-        )
-
-        expected_vars = {
-            "PREFECT__CLOUD__API": "api",
-            "PREFECT__CLOUD__AUTH_TOKEN": "token",
-            "PREFECT__CLOUD__AGENT__LABELS": str(
-                [
-                    socket.gethostname(),
-                    "azure-flow-storage",
-                    "gcs-flow-storage",
-                    "s3-flow-storage",
-                    "github-flow-storage",
-                    "webhook-flow-storage",
-                    "gitlab-flow-storage",
-                ]
-            ),
+            "PREFECT__CLOUD__API": "https://api.prefect.io",
+            "PREFECT__CLOUD__AUTH_TOKEN": "TEST_TOKEN",
+            "PREFECT__CLOUD__AGENT__LABELS": str(DEFAULT_AGENT_LABELS),
             "PREFECT__CONTEXT__FLOW_RUN_ID": "id",
             "PREFECT__CONTEXT__FLOW_ID": "foo",
             "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
@@ -188,170 +110,134 @@ def test_populate_env_vars(cloud_api):
             "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
             "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
         }
+    )
 
-        assert env_vars == expected_vars
+    assert env_vars == expected
 
 
-def test_populate_env_vars_includes_agent_labels(cloud_api):
-    with set_temporary_config(
-        {
-            "cloud.api": "api",
-            "logging.log_to_cloud": True,
-            "cloud.agent.auth_token": "token",
-        }
-    ):
-        agent = LocalAgent(labels=["42", "marvin"])
+@pytest.mark.parametrize("flag", [True, False])
+def test_populate_env_vars_sets_log_to_cloud(flag):
+    agent = LocalAgent(no_cloud_logs=flag)
+    assert agent.log_to_cloud is not flag
+    env_vars = agent.populate_env_vars(
+        GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
+    )
+    assert env_vars["PREFECT__LOGGING__LOG_TO_CLOUD"] == str(not flag).lower()
 
-        env_vars = agent.populate_env_vars(
-            GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
-        )
 
-        expected_vars = {
-            "PREFECT__CLOUD__API": "api",
-            "PREFECT__CLOUD__AUTH_TOKEN": "token",
-            "PREFECT__CLOUD__AGENT__LABELS": str(
-                [
-                    "42",
-                    "marvin",
-                    socket.gethostname(),
-                    "azure-flow-storage",
-                    "gcs-flow-storage",
-                    "s3-flow-storage",
-                    "github-flow-storage",
-                    "webhook-flow-storage",
-                    "gitlab-flow-storage",
-                ]
-            ),
-            "PREFECT__CONTEXT__FLOW_RUN_ID": "id",
-            "PREFECT__CONTEXT__FLOW_ID": "foo",
-            "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
-            "PREFECT__LOGGING__LOG_TO_CLOUD": "true",
-            "PREFECT__LOGGING__LEVEL": "INFO",
-            "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
-            "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
-        }
+def test_populate_env_vars_from_agent_config():
+    agent = LocalAgent(env_vars=dict(AUTH_THING="foo"))
 
-        assert env_vars == expected_vars
+    env_vars = agent.populate_env_vars(
+        GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
+    )
 
+    assert env_vars["AUTH_THING"] == "foo"
 
-def test_local_agent_deploy_processes_local_storage(monkeypatch, cloud_api):
 
-    popen = MagicMock()
-    monkeypatch.setattr("prefect.agent.local.agent.Popen", popen)
+def test_populate_env_vars_removes_none_values():
+    agent = LocalAgent(env_vars=dict(MISSING_VAR=None))
 
-    agent = LocalAgent()
-    agent.deploy_flow(
-        flow_run=GraphQLResult(
-            {
-                "flow": GraphQLResult(
-                    {
-                        "storage": Local(directory="test").serialize(),
-                        "id": "foo",
-                        "core_version": "0.13.0",
-                    }
-                ),
-                "id": "id",
-            }
-        )
+    env_vars = agent.populate_env_vars(
+        GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
     )
 
-    assert popen.called
-    assert len(agent.processes) == 1
+    assert "MISSING_VAR" not in env_vars
 
 
-def test_local_agent_deploy_processes_gcs_storage(monkeypatch, cloud_api):
+def test_populate_env_vars_includes_agent_labels():
+    agent = LocalAgent(labels=["42", "marvin"])
 
-    popen = MagicMock()
-    monkeypatch.setattr("prefect.agent.local.agent.Popen", popen)
-
-    agent = LocalAgent()
-    agent.deploy_flow(
-        flow_run=GraphQLResult(
-            {
-                "flow": GraphQLResult(
-                    {
-                        "storage": GCS(bucket="test").serialize(),
-                        "id": "foo",
-                        "core_version": "0.13.0",
-                    }
-                ),
-                "id": "id",
-            }
-        )
+    env_vars = agent.populate_env_vars(
+        GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
     )
+    expected = str(["42", "marvin"] + DEFAULT_AGENT_LABELS)
+    assert env_vars["PREFECT__CLOUD__AGENT__LABELS"] == expected
 
-    assert popen.called
-    assert len(agent.processes) == 1
 
+def test_populate_env_vars_import_paths():
+    agent = LocalAgent(import_paths=["paths"])
+    env_vars = agent.populate_env_vars(
+        GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
+    )
+    assert "paths" in env_vars["PYTHONPATH"]
 
-def test_local_agent_deploy_processes_s3_storage(monkeypatch, cloud_api):
 
-    popen = MagicMock()
-    monkeypatch.setattr("prefect.agent.local.agent.Popen", popen)
+def test_populate_env_vars_keep_existing_python_path(monkeypatch):
+    monkeypatch.setenv("PYTHONPATH", "cool:python:path")
 
-    agent = LocalAgent()
-    agent.deploy_flow(
-        flow_run=GraphQLResult(
-            {
-                "flow": GraphQLResult(
-                    {
-                        "storage": GCS(bucket="test").serialize(),
-                        "id": "foo",
-                        "core_version": "0.13.0",
-                    }
-                ),
-                "id": "id",
-            }
-        )
+    agent = LocalAgent(import_paths=["paths"])
+    env_vars = agent.populate_env_vars(
+        GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
     )
 
-    assert popen.called
-    assert len(agent.processes) == 1
+    python_path = env_vars["PYTHONPATH"]
+    assert "cool:python:path" in python_path
+    assert "paths" in python_path
 
 
-def test_local_agent_deploy_processes_azure_storage(monkeypatch, cloud_api):
+def test_populate_env_vars_no_existing_python_path(monkeypatch):
+    monkeypatch.delenv("PYTHONPATH", raising=False)
 
-    popen = MagicMock()
-    monkeypatch.setattr("prefect.agent.local.agent.Popen", popen)
+    agent = LocalAgent(import_paths=["paths"])
+    env_vars = agent.populate_env_vars(
+        GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}})
+    )
+    assert "paths" in env_vars["PYTHONPATH"]
 
-    agent = LocalAgent()
-    agent.deploy_flow(
-        flow_run=GraphQLResult(
+
+def test_populate_env_vars_from_run_config(tmpdir):
+    agent = LocalAgent(env_vars={"KEY1": "VAL1", "KEY2": "VAL2"})
+    working_dir = str(tmpdir)
+
+    run = LocalRun(
+        env={"KEY2": "OVERRIDE", "PREFECT__LOGGING__LEVEL": "TEST"},
+        working_dir=working_dir,
+    )
+
+    env_vars = agent.populate_env_vars(
+        GraphQLResult(
             {
-                "flow": GraphQLResult(
-                    {
-                        "storage": GCS(bucket="test").serialize(),
-                        "id": "foo",
-                        "core_version": "0.13.0",
-                    }
-                ),
                 "id": "id",
+                "name": "name",
+                "flow": {"id": "foo", "run_config": run.serialize()},
             }
-        )
+        ),
+        run,
     )
+    assert env_vars["KEY1"] == "VAL1"
+    assert env_vars["KEY2"] == "OVERRIDE"
+    assert env_vars["PREFECT__LOGGING__LEVEL"] == "TEST"
+    assert working_dir in env_vars["PYTHONPATH"]
 
-    assert popen.called
-    assert len(agent.processes) == 1
-
-
-def test_local_agent_deploy_processes_webhook_storage(monkeypatch, runner_token):
 
+@pytest.mark.parametrize(
+    "storage",
+    [
+        Local(directory="test"),
+        GCS(bucket="test"),
+        S3(bucket="test"),
+        Azure(container="test"),
+        GitLab("test/repo", path="path/to/flow.py"),
+        Webhook(
+            build_request_kwargs={"url": "test-service/upload"},
+            build_request_http_method="POST",
+            get_flow_request_kwargs={"url": "test-service/download"},
+            get_flow_request_http_method="GET",
+        ),
+    ],
+)
+def test_local_agent_deploy_processes_valid_storage(storage, monkeypatch):
     popen = MagicMock()
     monkeypatch.setattr("prefect.agent.local.agent.Popen", popen)
 
     agent = LocalAgent()
-    webhook = Webhook(
-        build_request_kwargs={"url": "test-service/upload"},
-        build_request_http_method="POST",
-        get_flow_request_kwargs={"url": "test-service/download"},
-        get_flow_request_http_method="GET",
-    )
     agent.deploy_flow(
         flow_run=GraphQLResult(
             {
                 "flow": GraphQLResult(
                     {
-                        "storage": webhook.serialize(),
+                        "storage": storage.serialize(),
                         "id": "foo",
                         "core_version": "0.13.0",
                     }
@@ -365,16 +251,13 @@ def test_local_agent_deploy_processes_webhook_storage(monkeypatch, runner_token)
     assert len(agent.processes) == 1
 
 
-def test_local_agent_deploy_storage_raises_not_supported_storage(
-    monkeypatch, cloud_api
-):
-
+def test_local_agent_deploy_raises_unsupported_storage(monkeypatch):
     popen = MagicMock()
     monkeypatch.setattr("prefect.agent.local.agent.Popen", popen)
 
     agent = LocalAgent()
 
-    with pytest.raises(ValueError):
+    with pytest.raises(TypeError, match="Unsupported Storage type: Docker"):
         agent.deploy_flow(
             flow_run=GraphQLResult(
                 {
@@ -392,8 +275,7 @@ def test_local_agent_deploy_storage_raises_not_supported_storage(
     assert len(agent.processes) == 0
 
 
-def test_local_agent_deploy_storage_fails_none(monkeypatch, cloud_api):
-
+def test_local_agent_deploy_storage_fails_none(monkeypatch):
     client = MagicMock()
     set_state = MagicMock()
     client.return_value.set_flow_run_state = set_state
@@ -423,89 +305,88 @@ def test_local_agent_deploy_storage_fails_none(monkeypatch, cloud_api):
     assert client.called
 
 
-def test_local_agent_deploy_import_paths(monkeypatch, cloud_api):
+def test_local_agent_deploy_unsupported_run_config(monkeypatch):
     popen = MagicMock()
     monkeypatch.setattr("prefect.agent.local.agent.Popen", popen)
 
-    agent = LocalAgent(import_paths=["paths"])
-    agent.deploy_flow(
-        flow_run=GraphQLResult(
-            {
-                "flow": GraphQLResult(
-                    {
-                        "storage": Local(directory="test").serialize(),
+    agent = LocalAgent()
+
+    with pytest.raises(TypeError, match="Unsupported RunConfig type: Kubernetes"):
+        agent.deploy_flow(
+            flow_run=GraphQLResult(
+                {
+                    "id": "id",
+                    "flow": {
+                        "storage": Local().serialize(),
+                        "run_config": KubernetesRun().serialize(),
                         "id": "foo",
                         "core_version": "0.13.0",
-                    }
-                ),
-                "id": "id",
-            }
+                    },
+                },
+            )
         )
-    )
-
-    assert popen.called
-    assert len(agent.processes) == 1
-    assert "paths" in popen.call_args[1]["env"]["PYTHONPATH"]
 
+    assert not popen.called
+    assert len(agent.processes) == 0
 
-def test_local_agent_deploy_keep_existing_python_path(monkeypatch, cloud_api):
-    monkeypatch.setenv("PYTHONPATH", "cool:python:path")
 
+@pytest.mark.parametrize("working_dir", [None, "existing"])
+def test_local_agent_deploy_run_config_working_dir(monkeypatch, working_dir, tmpdir):
     popen = MagicMock()
     monkeypatch.setattr("prefect.agent.local.agent.Popen", popen)
 
-    agent = LocalAgent(import_paths=["paths"])
+    if working_dir is not None:
+        working_dir = str(tmpdir)
+
+    agent = LocalAgent()
+
     agent.deploy_flow(
         flow_run=GraphQLResult(
             {
-                "flow": GraphQLResult(
-                    {
-                        "storage": Local(directory="test").serialize(),
-                        "id": "foo",
-                        "core_version": "0.13.0",
-                    }
-                ),
                 "id": "id",
-            }
+                "flow": {
+                    "storage": Local().serialize(),
+                    "run_config": LocalRun(working_dir=working_dir).serialize(),
+                    "id": "foo",
+                    "core_version": "0.13.0",
+                },
+            },
         )
     )
 
-    python_path = popen.call_args[1]["env"]["PYTHONPATH"]
-
     assert popen.called
     assert len(agent.processes) == 1
-    assert "cool:python:path" in python_path
-    assert "paths" in python_path
+    assert popen.call_args[1]["cwd"] == working_dir
 
 
-def test_local_agent_deploy_no_existing_python_path(monkeypatch, cloud_api):
-    monkeypatch.delenv("PYTHONPATH", raising=False)
-
+def test_local_agent_deploy_run_config_missing_working_dir(monkeypatch, tmpdir):
     popen = MagicMock()
     monkeypatch.setattr("prefect.agent.local.agent.Popen", popen)
 
-    agent = LocalAgent(import_paths=["paths"])
-    agent.deploy_flow(
-        flow_run=GraphQLResult(
-            {
-                "flow": GraphQLResult(
-                    {
-                        "storage": Local(directory="test").serialize(),
+    working_dir = str(tmpdir.join("missing"))
+
+    agent = LocalAgent()
+
+    with pytest.raises(ValueError, match="nonexistent `working_dir`"):
+        agent.deploy_flow(
+            flow_run=GraphQLResult(
+                {
+                    "id": "id",
+                    "flow": {
+                        "storage": Local().serialize(),
+                        "run_config": LocalRun(working_dir=working_dir).serialize(),
                         "id": "foo",
                         "core_version": "0.13.0",
-                    }
-                ),
-                "id": "id",
-            }
+                    },
+                },
+            )
         )
-    )
 
-    assert popen.called
-    assert len(agent.processes) == 1
-    assert "paths" in popen.call_args[1]["env"]["PYTHONPATH"]
+    assert not popen.called
+    assert not agent.processes
 
 
-def test_generate_supervisor_conf(cloud_api):
+def test_generate_supervisor_conf():
     agent = LocalAgent()
 
     conf = agent.generate_supervisor_conf(
@@ -529,9 +410,7 @@ def test_generate_supervisor_conf(cloud_api):
         (1, True, (("agent", "INFO", "Process PID 1234 returned non-zero exit code"),)),
     ),
 )
-def test_local_agent_heartbeat(
-    monkeypatch, cloud_api, returncode, show_flow_logs, logs
-):
+def test_local_agent_heartbeat(monkeypatch, returncode, show_flow_logs, logs):
     popen = MockPopen()
     # expect a process to be called with the following command (with specified behavior)
     popen.set_command(
@@ -589,27 +468,8 @@ def test_local_agent_heartbeat(
     assert len(agent.processes) == 0
 
 
-def test_local_agent_start_max_polls(monkeypatch, runner_token, cloud_api):
-    on_shutdown = MagicMock()
-    monkeypatch.setattr("prefect.agent.local.agent.LocalAgent.on_shutdown", on_shutdown)
-
-    agent_process = MagicMock()
-    monkeypatch.setattr("prefect.agent.agent.Agent.agent_process", agent_process)
-
-    agent_connect = MagicMock(return_value="id")
-    monkeypatch.setattr("prefect.agent.agent.Agent.agent_connect", agent_connect)
-
-    heartbeat = MagicMock()
-    monkeypatch.setattr("prefect.agent.local.agent.LocalAgent.heartbeat", heartbeat)
-
-    agent = LocalAgent(max_polls=1)
-    agent.start()
-
-    assert agent_process.called
-    assert heartbeat.called
-
-
-def test_local_gent_start_max_polls_count(monkeypatch, runner_token, cloud_api):
+@pytest.mark.parametrize("max_polls", [0, 1, 2])
+def test_local_agent_start_max_polls(max_polls, monkeypatch, runner_token):
     on_shutdown = MagicMock()
     monkeypatch.setattr("prefect.agent.local.agent.LocalAgent.on_shutdown", on_shutdown)
 
@@ -622,54 +482,10 @@ def test_local_gent_start_max_polls_count(monkeypatch, runner_token, cloud_api):
     heartbeat = MagicMock()
     monkeypatch.setattr("prefect.agent.local.agent.LocalAgent.heartbeat", heartbeat)
 
-    agent = LocalAgent(max_polls=2)
+    agent = LocalAgent(max_polls=max_polls)
     agent.start()
 
-    assert on_shutdown.call_count == 1
-    assert agent_process.call_count == 2
+    assert agent_connect.call_count == 1
+    assert agent_process.call_count == max_polls
     assert heartbeat.call_count == 1
-
-
-def test_local_agent_start_max_polls_zero(monkeypatch, runner_token, cloud_api):
-    on_shutdown = MagicMock()
-    monkeypatch.setattr("prefect.agent.local.agent.LocalAgent.on_shutdown", on_shutdown)
-
-    agent_process = MagicMock()
-    monkeypatch.setattr("prefect.agent.agent.Agent.agent_process", agent_process)
-
-    agent_connect = MagicMock(return_value="id")
-    monkeypatch.setattr("prefect.agent.agent.Agent.agent_connect", agent_connect)
-
-    heartbeat = MagicMock()
-    monkeypatch.setattr("prefect.agent.local.agent.LocalAgent.heartbeat", heartbeat)
-
-    agent = LocalAgent(max_polls=0)
-    agent.start()
-
     assert on_shutdown.call_count == 1
-    assert agent_process.call_count == 0
-    assert heartbeat.call_count == 1
-
-
-def test_local_agent_deploy_processes_gitlab_storage(monkeypatch, cloud_api):
-    popen = MagicMock()
-    monkeypatch.setattr("prefect.agent.local.agent.Popen", popen)
-
-    agent = LocalAgent()
-    gitlab = GitLab("test/repo", path="path/to/flow.py")
-    agent.deploy_flow(
-        flow_run=GraphQLResult(
-            {
-                "flow": GraphQLResult(
-                    {
-                        "storage": gitlab.serialize(),
-                        "id": "foo",
-                    }
-                ),
-                "id": "id",
-            }
-        )
-    )
-
-    assert popen.called
-    assert len(agent.processes) == 1
diff --git a/tests/core/test_flow.py b/tests/core/test_flow.py
index 6b763a07ae..ac0d1c0f4f 100644
--- a/tests/core/test_flow.py
+++ b/tests/core/test_flow.py
@@ -41,6 +41,8 @@ from prefect.engine.state import (
     TriggerFailed,
     TimedOut,
 )
+from prefect.environments.execution import LocalEnvironment
+from prefect.run_configs import LocalRun
 from prefect.schedules.clocks import ClockEvent
 from prefect.tasks.core.function import FunctionTask
 from prefect.utilities.configuration import set_temporary_config
@@ -2579,6 +2581,7 @@ class TestFlowRegister:
         assert f.storage.image_tag == "BIG"
         assert f.environment.labels == set()
 
+    @pytest.mark.parametrize("kind", ["environment", "run_config"])
     @pytest.mark.parametrize(
         "storage",
         [
@@ -2588,15 +2591,19 @@ class TestFlowRegister:
             prefect.environments.storage.Azure(container="windows"),
         ],
     )
-    def test_flow_register_auto_labels_environment_if_labeled_storage_used(
-        self, monkeypatch, storage
+    def test_flow_register_auto_labels_if_labeled_storage_used(
+        self, monkeypatch, storage, kind
     ):
         monkeypatch.setattr("prefect.Client", MagicMock())
         f = Flow(name="Test me!! I should get labeled", storage=storage)
+        if kind == "run_config":
+            obj = f.run_config = LocalRun(labels=["test-label"])
+        else:
+            obj = f.environment = LocalEnvironment(labels=["test-label"])
 
         f.register("My-project", build=False)
 
-        assert len(f.environment.labels) == 1
+        assert obj.labels == {"test-label", *storage.labels}
 
     @pytest.mark.parametrize(
         "storage",
@@ -3050,16 +3057,17 @@ def test_results_write_to_custom_formatters(tmpdir):
     }
 
 
-def test_run_agent_passes_environment_labels(monkeypatch):
+@pytest.mark.parametrize("kind", ["environment", "run_config"])
+def test_run_agent_passes_flow_labels(monkeypatch, kind):
     agent = MagicMock()
     monkeypatch.setattr("prefect.agent.local.LocalAgent", agent)
+    labels = ["test", "test", "test2"]
 
-    f = Flow(
-        "test",
-        environment=prefect.environments.LocalEnvironment(
-            labels=["test", "test", "test2"]
-        ),
-    )
+    f = Flow("test")
+    if kind == "run_config":
+        f.run_config = LocalRun(labels=labels)
+    else:
+        f.environment = LocalEnvironment(labels=labels)
     f.run_agent()
 
     assert type(agent.call_args[1]["labels"]) is list
diff --git a/tests/engine/test_task_runner.py b/tests/engine/test_task_runner.py
index e02639c14d..e7a3ede10a 100644
--- a/tests/engine/test_task_runner.py
+++ b/tests/engine/test_task_runner.py
@@ -307,6 +307,9 @@ def test_task_runner_accepts_dictionary_of_edges():
     sys.platform == "win32", reason="Windows doesn't support any timeout logic"
 )
 def test_timeout_actually_stops_execution():
+    # Note this replicates test coverage from
+    # `tests.core.test_flow.test_timeout_actually_stops_execution`
+
     # Note: this is a potentially brittle test! In some cases (local and sync) signal.alarm
     # is used as the mechanism for timing out a task. This passes off the job of measuring
     # the time for the timeout to the OS, which uses the "wallclock" as reference (the real
@@ -2126,7 +2129,7 @@ class TestLooping:
         logs = [
             log
             for log in caplog.records
-            if "TaskRunner" in log.name and "finished" in log.message
+            if "TaskRunner" in log.name and "Finished" in log.message
         ]
         assert len(logs) >= 1  # a finished log was in fact created
         assert len(logs) <= 2  # but not too many were issued
diff --git a/tests/run_configs/test_docker.py b/tests/run_configs/test_docker.py
new file mode 100644
index 0000000000..6d96f1d3d0
--- /dev/null
+++ b/tests/run_configs/test_docker.py
@@ -0,0 +1,20 @@
+from prefect.run_configs import DockerRun
+
+
+def test_no_args():
+    config = DockerRun()
+    assert config.env is None
+    assert config.image is None
+    assert config.labels == set()
+
+
+def test_all_args(tmpdir):
+    working_dir = str(tmpdir)
+    config = DockerRun(
+        env={"hello": "world"},
+        image="testing",
+        labels=["a", "b"],
+    )
+    assert config.env == {"hello": "world"}
+    assert config.image == "testing"
+    assert config.labels == {"a", "b"}
diff --git a/tests/run_configs/test_local.py b/tests/run_configs/test_local.py
new file mode 100644
index 0000000000..8ea9554325
--- /dev/null
+++ b/tests/run_configs/test_local.py
@@ -0,0 +1,29 @@
+import os
+
+from prefect.run_configs import LocalRun
+
+
+def test_no_args():
+    config = LocalRun()
+    assert config.env is None
+    assert config.working_dir is None
+    assert config.labels == set()
+
+
+def test_all_args(tmpdir):
+    working_dir = str(tmpdir)
+    config = LocalRun(
+        env={"hello": "world"},
+        working_dir=working_dir,
+        labels=["a", "b"],
+    )
+    assert config.env == {"hello": "world"}
+    assert config.working_dir == working_dir
+    assert config.labels == {"a", "b"}
+
+
+def test_working_dir_relpath_to_abspath():
+    relpath = os.path.join("local", "path")
+    abspath = os.path.abspath(relpath)
+    config = LocalRun(working_dir=relpath)
+    assert config.working_dir == abspath
diff --git a/tests/serialization/test_run_configs.py b/tests/serialization/test_run_configs.py
index 9f62ba8fc4..7c59deef48 100644
--- a/tests/serialization/test_run_configs.py
+++ b/tests/serialization/test_run_configs.py
@@ -1,8 +1,6 @@
 import pytest
 
-from marshmallow import ValidationError
-
-from prefect.run_configs import KubernetesRun
+from prefect.run_configs import KubernetesRun, LocalRun, DockerRun
 from prefect.serialization.run_config import RunConfigSchema
 
 
@@ -29,7 +27,7 @@ from prefect.serialization.run_config import RunConfigSchema
         ),
     ],
 )
-def test_serialize_kubernetes_job(config):
+def test_serialize_kubernetes_run(config):
     msg = RunConfigSchema().dump(config)
     config2 = RunConfigSchema().load(msg)
     assert sorted(config.labels) == sorted(config2.labels)
@@ -45,3 +43,43 @@ def test_serialize_kubernetes_job(config):
     ]
     for field in fields:
         assert getattr(config, field) == getattr(config2, field)
+
+
+@pytest.mark.parametrize(
+    "config",
+    [
+        LocalRun(),
+        LocalRun(
+            env={"test": "foo"},
+            working_dir="/path/to/dir",
+            labels=["a", "b"],
+        ),
+    ],
+)
+def test_serialize_local_run(config):
+    msg = RunConfigSchema().dump(config)
+    config2 = RunConfigSchema().load(msg)
+    assert sorted(config.labels) == sorted(config2.labels)
+    fields = ["env", "working_dir"]
+    for field in fields:
+        assert getattr(config, field) == getattr(config2, field)
+
+
+@pytest.mark.parametrize(
+    "config",
+    [
+        DockerRun(),
+        DockerRun(
+            env={"test": "foo"},
+            image="testing",
+            labels=["a", "b"],
+        ),
+    ],
+)
+def test_serialize_docker_run(config):
+    msg = RunConfigSchema().dump(config)
+    config2 = RunConfigSchema().load(msg)
+    assert sorted(config.labels) == sorted(config2.labels)
+    fields = ["env", "image"]
+    for field in fields:
+        assert getattr(config, field) == getattr(config2, field)
diff --git a/tests/utilities/test_executors.py b/tests/utilities/test_executors.py
index 0aacc976ca..4ca258d000 100644
--- a/tests/utilities/test_executors.py
+++ b/tests/utilities/test_executors.py
@@ -10,113 +10,180 @@ import pytest
 import prefect
 from prefect.utilities.configuration import set_temporary_config
 from prefect.utilities.executors import (
-    timeout_handler,
+    run_with_thread_timeout,
+    run_with_multiprocess_timeout,
     tail_recursive,
     RecursiveCall,
 )
 
 
-def test_timeout_handler_times_out():
+# We will test the low-level timeout handlers here and `run_task_with_timeout`
+# is covered in `tests.core.test_flow.test_timeout_actually_stops_execution`
+# and `tests.engine.test_task_runner.test_timeout_actually_stops_execution`
+TIMEOUT_HANDLERS = [run_with_thread_timeout, run_with_multiprocess_timeout]
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="Windows doesn't support any timeout logic"
+)
+@pytest.mark.parametrize("timeout_handler", TIMEOUT_HANDLERS)
+def test_timeout_handler_times_out(timeout_handler):
     slow_fn = lambda: time.sleep(2)
     with pytest.raises(TimeoutError):
-        timeout_handler(slow_fn, timeout=1)
+        timeout_handler(slow_fn, timeout=2)
 
 
 @pytest.mark.skipif(
     sys.platform == "win32", reason="Windows doesn't support any timeout logic"
 )
-def test_timeout_handler_actually_stops_execution():
+@pytest.mark.parametrize("timeout_handler", TIMEOUT_HANDLERS)
+def test_timeout_handler_actually_stops_execution(timeout_handler):
     with tempfile.TemporaryDirectory() as call_dir:
         FILE = os.path.join(call_dir, "test.txt")
+        # Increase timeout for multiprocess because its startup time is slower and
+        # the file was not always created
+        TIMEOUT = 1 if timeout_handler is run_with_thread_timeout else 3
+        WRITES = 6
 
         def slow_fn():
-            "Runs for 1.5 seconds, writes to file 6 times"
-            iters = 0
-            while iters < 6:
-                time.sleep(0.26)
+            "Runs for TIMEOUT * 2 seconds, writes to file WRITES times"
+            for _ in range(WRITES):
                 with open(FILE, "a") as f:
                     f.write("called\n")
-                iters += 1
+                time.sleep((TIMEOUT * 2) / WRITES)
 
         with pytest.raises(TimeoutError):
-            # allow for at most 3 writes
-            timeout_handler(slow_fn, timeout=1)
+            # We should get less than WRITES lines -- roughly half expected
+            timeout_handler(slow_fn, timeout=TIMEOUT)
 
         time.sleep(0.5)
         with open(FILE, "r") as g:
             contents = g.read()
 
-    assert len(contents.split("\n")) <= 4
+    assert len(contents.split("\n")) < WRITES
 
 
-def test_timeout_handler_passes_args_and_kwargs_and_returns():
-    def do_nothing(x, y=None):
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="Windows doesn't support any timeout logic"
+)
+@pytest.mark.parametrize("timeout_handler", TIMEOUT_HANDLERS)
+def test_timeout_handler_passes_args_and_kwargs_and_returns(timeout_handler):
+    def just_return(x, y=None):
         return x, y
 
-    assert timeout_handler(do_nothing, 5, timeout=1, y="yellow") == (5, "yellow")
+    assert timeout_handler(
+        just_return, args=[5], kwargs=dict(y="yellow"), timeout=2
+    ) == (
+        5,
+        "yellow",
+    )
 
 
-def test_timeout_handler_doesnt_swallow_bad_args():
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="Windows doesn't support any timeout logic"
+)
+@pytest.mark.parametrize("timeout_handler", TIMEOUT_HANDLERS)
+def test_timeout_handler_doesnt_swallow_bad_args(timeout_handler):
     def do_nothing(x, y=None):
         return x, y
 
     with pytest.raises(TypeError):
-        timeout_handler(do_nothing, timeout=1)
+        timeout_handler(do_nothing, timeout=2)
 
     with pytest.raises(TypeError):
-        timeout_handler(do_nothing, 5, timeout=1, z=10)
+        timeout_handler(do_nothing, args=[5], kwargs=dict(z=10), timeout=2)
 
     with pytest.raises(TypeError):
-        timeout_handler(do_nothing, 5, timeout=1, y="s", z=10)
+        timeout_handler(do_nothing, args=[5], kwargs=dict(y="s", z=10), timeout=2)
 
 
-def test_timeout_handler_reraises():
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="Windows doesn't support any timeout logic"
+)
+@pytest.mark.parametrize("timeout_handler", TIMEOUT_HANDLERS)
+def test_timeout_handler_reraises(timeout_handler):
     def do_something():
         raise ValueError("test")
 
     with pytest.raises(ValueError, match="test"):
-        timeout_handler(do_something, timeout=1)
+        timeout_handler(do_something, timeout=2)
 
 
-@pytest.mark.skipif(sys.platform == "win32", reason="Test fails on Windows")
-def test_timeout_handler_allows_function_to_spawn_new_process():
+# Define a top-level helper function for a null-op process target, must be defined
+# as a non-local for the python native pickler used within `my_process`
+def do_nothing():
+    return None
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="Windows doesn't support any timeout logic"
+)
+@pytest.mark.parametrize("timeout_handler", TIMEOUT_HANDLERS)
+def test_timeout_handler_allows_function_to_spawn_new_process(timeout_handler):
     def my_process():
-        p = multiprocessing.Process(target=lambda: 5)
+        p = multiprocessing.Process(target=do_nothing())
         p.start()
         p.join()
         p.terminate()
 
-    assert timeout_handler(my_process, timeout=1) is None
+    assert timeout_handler(my_process, timeout=3) is None
 
 
-@pytest.mark.skipif(sys.platform == "win32", reason="Test fails on Windows")
-def test_timeout_handler_allows_function_to_spawn_new_thread():
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="Windows doesn't support any timeout logic"
+)
+@pytest.mark.parametrize("timeout_handler", TIMEOUT_HANDLERS)
+def test_timeout_handler_allows_function_to_spawn_new_thread(timeout_handler):
     def my_thread():
         t = threading.Thread(target=lambda: 5)
         t.start()
         t.join()
 
-    assert timeout_handler(my_thread, timeout=1) is None
+    assert timeout_handler(my_thread, timeout=3) is None
 
 
-def test_timeout_handler_doesnt_do_anything_if_no_timeout(monkeypatch):
-    assert timeout_handler(lambda: 4, timeout=1) == 4
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="Windows doesn't support any timeout logic"
+)
+@pytest.mark.parametrize("timeout_handler", TIMEOUT_HANDLERS)
+def test_timeout_handler_doesnt_do_anything_if_no_timeout(timeout_handler):
+    assert timeout_handler(lambda: 4, timeout=2) == 4
     assert timeout_handler(lambda: 4) == 4
 
 
-def test_timeout_handler_preserves_context():
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="Windows doesn't support any timeout logic"
+)
+@pytest.mark.parametrize("timeout_handler", TIMEOUT_HANDLERS)
+def test_timeout_handler_preserves_context(timeout_handler):
     def my_fun(x, **kwargs):
         return prefect.context.get("test_key")
 
     with prefect.context(test_key=42):
-        res = timeout_handler(my_fun, 2, timeout=1)
+        res = timeout_handler(my_fun, args=[2], timeout=2)
 
     assert res == 42
 
 
-def test_timeout_handler_preserves_logging(caplog):
-    timeout_handler(prefect.Flow("logs").run, timeout=2)
-    assert len(caplog.records) >= 2  # 1 INFO to start, 1 INFO to end
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="Windows doesn't support any timeout logic"
+)
+def test_run_with_thread_timeout_preserves_logging(caplog):
+    run_with_thread_timeout(prefect.Flow("logs").run, timeout=2)
+    assert len(caplog.messages) >= 2  # 1 INFO to start, 1 INFO to end
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="Windows doesn't support any timeout logic"
+)
+def test_run_with_multiprocess_timeout_preserves_logging(capfd):
+    """
+    Requires fd capturing because the subprocess output won't be captured by caplog
+    """
+    run_with_multiprocess_timeout(prefect.Flow("logs").run, timeout=2)
+    stdout = capfd.readouterr().out
+    assert "Beginning Flow run" in stdout
+    assert "Flow run SUCCESS" in stdout
 
 
 def test_recursion_go_case():