We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
When a flow has no storage attached, the agent should mark the run as failed

It appears that if the Agent picks up a run with no storage, it will mark it as submitted. This run will then exist in limbo forever, as it can never be run. The agent should put it in a failed state if it can recognize it'll never run.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/prefect/agent/local/agent.py`
Content:
```
1 from sys import platform
2 from typing import Iterable
3 
4 import docker
5 
6 from prefect import config, context
7 from prefect.agent import Agent
8 from prefect.environments.storage import Docker
9 from prefect.serialization.storage import StorageSchema
10 from prefect.utilities.graphql import GraphQLResult
11 
12 
13 class LocalAgent(Agent):
14     """
15     Agent which deploys flow runs locally as Docker containers. Information on using the
16     Local Agent can be found at https://docs.prefect.io/cloud/agent/local.html
17 
18     Args:
19         - name (str, optional): An optional name to give this agent. Can also be set through
20             the environment variable `PREFECT__CLOUD__AGENT__NAME`. Defaults to "agent"
21         - labels (List[str], optional): a list of labels, which are arbitrary string identifiers used by Prefect
22             Agents when polling for work
23         - base_url (str, optional): URL for a Docker daemon server. Defaults to
24             `unix:///var/run/docker.sock` however other hosts such as
25             `tcp://0.0.0.0:2375` can be provided
26         - no_pull (bool, optional): Flag on whether or not to pull flow images.
27             Defaults to `False` if not provided here or in context.
28     """
29 
30     def __init__(
31         self,
32         name: str = None,
33         labels: Iterable[str] = None,
34         base_url: str = None,
35         no_pull: bool = None,
36     ) -> None:
37         super().__init__(name=name, labels=labels)
38 
39         if platform == "win32":
40             default_url = "npipe:////./pipe/docker_engine"
41         else:
42             default_url = "unix://var/run/docker.sock"
43         self.logger.debug(
44             "Platform {} and default docker daemon {}".format(platform, default_url)
45         )
46 
47         # Determine Daemon URL
48         self.base_url = base_url or context.get("base_url", default_url)
49         self.logger.debug("Base docker daemon url {}".format(self.base_url))
50 
51         # Determine pull specification
52         self.no_pull = no_pull or context.get("no_pull", False)
53         self.logger.debug("no_pull set to {}".format(self.no_pull))
54 
55         self.docker_client = docker.APIClient(base_url=self.base_url, version="auto")
56 
57         # Ping Docker daemon for connection issues
58         try:
59             self.logger.debug("Pinging docker daemon")
60             self.docker_client.ping()
61         except Exception as exc:
62             self.logger.exception(
63                 "Issue connecting to the Docker daemon. Make sure it is running."
64             )
65             raise exc
66 
67     def deploy_flows(self, flow_runs: list) -> None:
68         """
69         Deploy flow runs on your local machine as Docker containers
70 
71         Args:
72             - flow_runs (list): A list of GraphQLResult flow run objects
73         """
74         for flow_run in flow_runs:
75             self.logger.info(
76                 "Deploying flow run {}".format(flow_run.id)  # type: ignore
77             )
78 
79             storage = StorageSchema().load(flow_run.flow.storage)
80             if not isinstance(StorageSchema().load(flow_run.flow.storage), Docker):
81                 self.logger.error(
82                     "Storage for flow run {} is not of type Docker.".format(flow_run.id)
83                 )
84                 continue
85 
86             env_vars = self.populate_env_vars(flow_run=flow_run)
87 
88             if not self.no_pull and storage.registry_url:
89                 self.logger.info("Pulling image {}...".format(storage.name))
90                 try:
91                     pull_output = self.docker_client.pull(
92                         storage.name, stream=True, decode=True
93                     )
94                     for line in pull_output:
95                         self.logger.debug(line)
96                     self.logger.info(
97                         "Successfully pulled image {}...".format(storage.name)
98                     )
99                 except docker.errors.APIError as exc:
100                     self.logger.error("Issue pulling image {}".format(storage.name))
101 
102             # Create a container
103             self.logger.debug("Creating Docker container {}".format(storage.name))
104             container = self.docker_client.create_container(
105                 storage.name, command="prefect execute cloud-flow", environment=env_vars
106             )
107 
108             # Start the container
109             self.logger.debug(
110                 "Starting Docker container with ID {}".format(container.get("Id"))
111             )
112             self.docker_client.start(container=container.get("Id"))
113 
114     def populate_env_vars(self, flow_run: GraphQLResult) -> dict:
115         """
116         Populate metadata and variables in the environment variables for a flow run
117 
118         Args:
119             - flow_run (GraphQLResult): A flow run object
120 
121         Returns:
122             - dict: a dictionary representing the populated environment variables
123         """
124         return {
125             "PREFECT__CLOUD__API": config.cloud.api,
126             "PREFECT__CLOUD__AUTH_TOKEN": config.cloud.agent.auth_token,
127             "PREFECT__CLOUD__AGENT__LABELS": str(self.labels),
128             "PREFECT__CONTEXT__FLOW_RUN_ID": flow_run.id,  # type: ignore
129             "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
130             "PREFECT__LOGGING__LOG_TO_CLOUD": "true",
131             "PREFECT__LOGGING__LEVEL": "DEBUG",
132             "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
133             "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
134         }
135 
136 
137 if __name__ == "__main__":
138     LocalAgent().start()
139
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/prefect/agent/local/agent.py b/src/prefect/agent/local/agent.py
--- a/src/prefect/agent/local/agent.py
+++ b/src/prefect/agent/local/agent.py
@@ -5,6 +5,7 @@
 
 from prefect import config, context
 from prefect.agent import Agent
+from prefect.engine.state import Failed
 from prefect.environments.storage import Docker
 from prefect.serialization.storage import StorageSchema
 from prefect.utilities.graphql import GraphQLResult
@@ -78,9 +79,14 @@
 
             storage = StorageSchema().load(flow_run.flow.storage)
             if not isinstance(StorageSchema().load(flow_run.flow.storage), Docker):
-                self.logger.error(
-                    "Storage for flow run {} is not of type Docker.".format(flow_run.id)
+                msg = "Storage for flow run {} is not of type Docker.".format(
+                    flow_run.id
                 )
+                state_msg = "Agent {} failed to run flow: ".format(self.name) + msg
+                self.client.set_flow_run_state(
+                    flow_run.id, version=flow_run.version, state=Failed(state_msg)
+                )
+                self.logger.error(msg)
                 continue
 
             env_vars = self.populate_env_vars(flow_run=flow_run)
@@ -97,7 +103,15 @@
                         "Successfully pulled image {}...".format(storage.name)
                     )
                 except docker.errors.APIError as exc:
-                    self.logger.error("Issue pulling image {}".format(storage.name))
+                    msg = "Issue pulling image {}".format(storage.name)
+                    state_msg = (
+                        "Agent {} failed to pull image for flow: ".format(self.name)
+                        + msg
+                    )
+                    self.client.set_flow_run_state(
+                        flow_run.id, version=flow_run.version, state=Failed(msg)
+                    )
+                    self.logger.error(msg)
 
             # Create a container
             self.logger.debug("Creating Docker container {}".format(storage.name))

{"golden_diff": "diff --git a/src/prefect/agent/local/agent.py b/src/prefect/agent/local/agent.py\n--- a/src/prefect/agent/local/agent.py\n+++ b/src/prefect/agent/local/agent.py\n@@ -5,6 +5,7 @@\n \n from prefect import config, context\n from prefect.agent import Agent\n+from prefect.engine.state import Failed\n from prefect.environments.storage import Docker\n from prefect.serialization.storage import StorageSchema\n from prefect.utilities.graphql import GraphQLResult\n@@ -78,9 +79,14 @@\n \n             storage = StorageSchema().load(flow_run.flow.storage)\n             if not isinstance(StorageSchema().load(flow_run.flow.storage), Docker):\n-                self.logger.error(\n-                    \"Storage for flow run {} is not of type Docker.\".format(flow_run.id)\n+                msg = \"Storage for flow run {} is not of type Docker.\".format(\n+                    flow_run.id\n                 )\n+                state_msg = \"Agent {} failed to run flow: \".format(self.name) + msg\n+                self.client.set_flow_run_state(\n+                    flow_run.id, version=flow_run.version, state=Failed(state_msg)\n+                )\n+                self.logger.error(msg)\n                 continue\n \n             env_vars = self.populate_env_vars(flow_run=flow_run)\n@@ -97,7 +103,15 @@\n                         \"Successfully pulled image {}...\".format(storage.name)\n                     )\n                 except docker.errors.APIError as exc:\n-                    self.logger.error(\"Issue pulling image {}\".format(storage.name))\n+                    msg = \"Issue pulling image {}\".format(storage.name)\n+                    state_msg = (\n+                        \"Agent {} failed to pull image for flow: \".format(self.name)\n+                        + msg\n+                    )\n+                    self.client.set_flow_run_state(\n+                        flow_run.id, version=flow_run.version, state=Failed(msg)\n+                    )\n+                    self.logger.error(msg)\n \n             # Create a container\n             self.logger.debug(\"Creating Docker container {}\".format(storage.name))\n", "issue": "When a flow has no storage attached, the agent should mark the run as failed\n\r\nIt appears that if the Agent picks up a run with no storage, it will mark it as submitted. This run will then exist in limbo forever, as it can never be run. The agent should put it in a failed state if it can recognize it'll never run.\n", "before_files": [{"content": "from sys import platform\nfrom typing import Iterable\n\nimport docker\n\nfrom prefect import config, context\nfrom prefect.agent import Agent\nfrom prefect.environments.storage import Docker\nfrom prefect.serialization.storage import StorageSchema\nfrom prefect.utilities.graphql import GraphQLResult\n\n\nclass LocalAgent(Agent):\n    \"\"\"\n    Agent which deploys flow runs locally as Docker containers. Information on using the\n    Local Agent can be found at https://docs.prefect.io/cloud/agent/local.html\n\n    Args:\n        - name (str, optional): An optional name to give this agent. Can also be set through\n            the environment variable `PREFECT__CLOUD__AGENT__NAME`. Defaults to \"agent\"\n        - labels (List[str], optional): a list of labels, which are arbitrary string identifiers used by Prefect\n            Agents when polling for work\n        - base_url (str, optional): URL for a Docker daemon server. Defaults to\n            `unix:///var/run/docker.sock` however other hosts such as\n            `tcp://0.0.0.0:2375` can be provided\n        - no_pull (bool, optional): Flag on whether or not to pull flow images.\n            Defaults to `False` if not provided here or in context.\n    \"\"\"\n\n    def __init__(\n        self,\n        name: str = None,\n        labels: Iterable[str] = None,\n        base_url: str = None,\n        no_pull: bool = None,\n    ) -> None:\n        super().__init__(name=name, labels=labels)\n\n        if platform == \"win32\":\n            default_url = \"npipe:////./pipe/docker_engine\"\n        else:\n            default_url = \"unix://var/run/docker.sock\"\n        self.logger.debug(\n            \"Platform {} and default docker daemon {}\".format(platform, default_url)\n        )\n\n        # Determine Daemon URL\n        self.base_url = base_url or context.get(\"base_url\", default_url)\n        self.logger.debug(\"Base docker daemon url {}\".format(self.base_url))\n\n        # Determine pull specification\n        self.no_pull = no_pull or context.get(\"no_pull\", False)\n        self.logger.debug(\"no_pull set to {}\".format(self.no_pull))\n\n        self.docker_client = docker.APIClient(base_url=self.base_url, version=\"auto\")\n\n        # Ping Docker daemon for connection issues\n        try:\n            self.logger.debug(\"Pinging docker daemon\")\n            self.docker_client.ping()\n        except Exception as exc:\n            self.logger.exception(\n                \"Issue connecting to the Docker daemon. Make sure it is running.\"\n            )\n            raise exc\n\n    def deploy_flows(self, flow_runs: list) -> None:\n        \"\"\"\n        Deploy flow runs on your local machine as Docker containers\n\n        Args:\n            - flow_runs (list): A list of GraphQLResult flow run objects\n        \"\"\"\n        for flow_run in flow_runs:\n            self.logger.info(\n                \"Deploying flow run {}\".format(flow_run.id)  # type: ignore\n            )\n\n            storage = StorageSchema().load(flow_run.flow.storage)\n            if not isinstance(StorageSchema().load(flow_run.flow.storage), Docker):\n                self.logger.error(\n                    \"Storage for flow run {} is not of type Docker.\".format(flow_run.id)\n                )\n                continue\n\n            env_vars = self.populate_env_vars(flow_run=flow_run)\n\n            if not self.no_pull and storage.registry_url:\n                self.logger.info(\"Pulling image {}...\".format(storage.name))\n                try:\n                    pull_output = self.docker_client.pull(\n                        storage.name, stream=True, decode=True\n                    )\n                    for line in pull_output:\n                        self.logger.debug(line)\n                    self.logger.info(\n                        \"Successfully pulled image {}...\".format(storage.name)\n                    )\n                except docker.errors.APIError as exc:\n                    self.logger.error(\"Issue pulling image {}\".format(storage.name))\n\n            # Create a container\n            self.logger.debug(\"Creating Docker container {}\".format(storage.name))\n            container = self.docker_client.create_container(\n                storage.name, command=\"prefect execute cloud-flow\", environment=env_vars\n            )\n\n            # Start the container\n            self.logger.debug(\n                \"Starting Docker container with ID {}\".format(container.get(\"Id\"))\n            )\n            self.docker_client.start(container=container.get(\"Id\"))\n\n    def populate_env_vars(self, flow_run: GraphQLResult) -> dict:\n        \"\"\"\n        Populate metadata and variables in the environment variables for a flow run\n\n        Args:\n            - flow_run (GraphQLResult): A flow run object\n\n        Returns:\n            - dict: a dictionary representing the populated environment variables\n        \"\"\"\n        return {\n            \"PREFECT__CLOUD__API\": config.cloud.api,\n            \"PREFECT__CLOUD__AUTH_TOKEN\": config.cloud.agent.auth_token,\n            \"PREFECT__CLOUD__AGENT__LABELS\": str(self.labels),\n            \"PREFECT__CONTEXT__FLOW_RUN_ID\": flow_run.id,  # type: ignore\n            \"PREFECT__CLOUD__USE_LOCAL_SECRETS\": \"false\",\n            \"PREFECT__LOGGING__LOG_TO_CLOUD\": \"true\",\n            \"PREFECT__LOGGING__LEVEL\": \"DEBUG\",\n            \"PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS\": \"prefect.engine.cloud.CloudFlowRunner\",\n            \"PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS\": \"prefect.engine.cloud.CloudTaskRunner\",\n        }\n\n\nif __name__ == \"__main__\":\n    LocalAgent().start()\n", "path": "src/prefect/agent/local/agent.py"}], "after_files": [{"content": "from sys import platform\nfrom typing import Iterable\n\nimport docker\n\nfrom prefect import config, context\nfrom prefect.agent import Agent\nfrom prefect.engine.state import Failed\nfrom prefect.environments.storage import Docker\nfrom prefect.serialization.storage import StorageSchema\nfrom prefect.utilities.graphql import GraphQLResult\n\n\nclass LocalAgent(Agent):\n    \"\"\"\n    Agent which deploys flow runs locally as Docker containers. Information on using the\n    Local Agent can be found at https://docs.prefect.io/cloud/agent/local.html\n\n    Args:\n        - name (str, optional): An optional name to give this agent. Can also be set through\n            the environment variable `PREFECT__CLOUD__AGENT__NAME`. Defaults to \"agent\"\n        - labels (List[str], optional): a list of labels, which are arbitrary string identifiers used by Prefect\n            Agents when polling for work\n        - base_url (str, optional): URL for a Docker daemon server. Defaults to\n            `unix:///var/run/docker.sock` however other hosts such as\n            `tcp://0.0.0.0:2375` can be provided\n        - no_pull (bool, optional): Flag on whether or not to pull flow images.\n            Defaults to `False` if not provided here or in context.\n    \"\"\"\n\n    def __init__(\n        self,\n        name: str = None,\n        labels: Iterable[str] = None,\n        base_url: str = None,\n        no_pull: bool = None,\n    ) -> None:\n        super().__init__(name=name, labels=labels)\n\n        if platform == \"win32\":\n            default_url = \"npipe:////./pipe/docker_engine\"\n        else:\n            default_url = \"unix://var/run/docker.sock\"\n        self.logger.debug(\n            \"Platform {} and default docker daemon {}\".format(platform, default_url)\n        )\n\n        # Determine Daemon URL\n        self.base_url = base_url or context.get(\"base_url\", default_url)\n        self.logger.debug(\"Base docker daemon url {}\".format(self.base_url))\n\n        # Determine pull specification\n        self.no_pull = no_pull or context.get(\"no_pull\", False)\n        self.logger.debug(\"no_pull set to {}\".format(self.no_pull))\n\n        self.docker_client = docker.APIClient(base_url=self.base_url, version=\"auto\")\n\n        # Ping Docker daemon for connection issues\n        try:\n            self.logger.debug(\"Pinging docker daemon\")\n            self.docker_client.ping()\n        except Exception as exc:\n            self.logger.exception(\n                \"Issue connecting to the Docker daemon. Make sure it is running.\"\n            )\n            raise exc\n\n    def deploy_flows(self, flow_runs: list) -> None:\n        \"\"\"\n        Deploy flow runs on your local machine as Docker containers\n\n        Args:\n            - flow_runs (list): A list of GraphQLResult flow run objects\n        \"\"\"\n        for flow_run in flow_runs:\n            self.logger.info(\n                \"Deploying flow run {}\".format(flow_run.id)  # type: ignore\n            )\n\n            storage = StorageSchema().load(flow_run.flow.storage)\n            if not isinstance(StorageSchema().load(flow_run.flow.storage), Docker):\n                msg = \"Storage for flow run {} is not of type Docker.\".format(\n                    flow_run.id\n                )\n                state_msg = \"Agent {} failed to run flow: \".format(self.name) + msg\n                self.client.set_flow_run_state(\n                    flow_run.id, version=flow_run.version, state=Failed(state_msg)\n                )\n                self.logger.error(msg)\n                continue\n\n            env_vars = self.populate_env_vars(flow_run=flow_run)\n\n            if not self.no_pull and storage.registry_url:\n                self.logger.info(\"Pulling image {}...\".format(storage.name))\n                try:\n                    pull_output = self.docker_client.pull(\n                        storage.name, stream=True, decode=True\n                    )\n                    for line in pull_output:\n                        self.logger.debug(line)\n                    self.logger.info(\n                        \"Successfully pulled image {}...\".format(storage.name)\n                    )\n                except docker.errors.APIError as exc:\n                    msg = \"Issue pulling image {}\".format(storage.name)\n                    state_msg = (\n                        \"Agent {} failed to pull image for flow: \".format(self.name)\n                        + msg\n                    )\n                    self.client.set_flow_run_state(\n                        flow_run.id, version=flow_run.version, state=Failed(msg)\n                    )\n                    self.logger.error(msg)\n\n            # Create a container\n            self.logger.debug(\"Creating Docker container {}\".format(storage.name))\n            container = self.docker_client.create_container(\n                storage.name, command=\"prefect execute cloud-flow\", environment=env_vars\n            )\n\n            # Start the container\n            self.logger.debug(\n                \"Starting Docker container with ID {}\".format(container.get(\"Id\"))\n            )\n            self.docker_client.start(container=container.get(\"Id\"))\n\n    def populate_env_vars(self, flow_run: GraphQLResult) -> dict:\n        \"\"\"\n        Populate metadata and variables in the environment variables for a flow run\n\n        Args:\n            - flow_run (GraphQLResult): A flow run object\n\n        Returns:\n            - dict: a dictionary representing the populated environment variables\n        \"\"\"\n        return {\n            \"PREFECT__CLOUD__API\": config.cloud.api,\n            \"PREFECT__CLOUD__AUTH_TOKEN\": config.cloud.agent.auth_token,\n            \"PREFECT__CLOUD__AGENT__LABELS\": str(self.labels),\n            \"PREFECT__CONTEXT__FLOW_RUN_ID\": flow_run.id,  # type: ignore\n            \"PREFECT__CLOUD__USE_LOCAL_SECRETS\": \"false\",\n            \"PREFECT__LOGGING__LOG_TO_CLOUD\": \"true\",\n            \"PREFECT__LOGGING__LEVEL\": \"DEBUG\",\n            \"PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS\": \"prefect.engine.cloud.CloudFlowRunner\",\n            \"PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS\": \"prefect.engine.cloud.CloudTaskRunner\",\n        }\n\n\nif __name__ == \"__main__\":\n    LocalAgent().start()\n", "path": "src/prefect/agent/local/agent.py"}]}